Expose app runtime info across adapters

This commit is contained in:
lda
2026-08-30 07:40:58 +07:00 Verified
parent acd412a50a
commit 16553b82c7
9 changed files with 131 additions and 12 deletions
+2
View File
@@ -7,6 +7,7 @@ from .events import (
event_from_history_output, event_from_history_output,
) )
from .models import ( from .models import (
AppInfo,
CallError, CallError,
CallEvent, CallEvent,
CallEventKind, CallEventKind,
@@ -28,6 +29,7 @@ from .utils import generate_good_names
__all__ = [ __all__ = [
"App", "App",
"AppInfo",
"CallError", "CallError",
"CallEvent", "CallEvent",
"CallEventKind", "CallEventKind",
+19 -1
View File
@@ -2,12 +2,14 @@ import contextlib
import io import io
import json import json
from dataclasses import asdict from dataclasses import asdict
from datetime import UTC, datetime
from time import monotonic
from typing import Literal from typing import Literal
from uuid import uuid4 from uuid import uuid4
from ipython_demo.utils import generate_good_names from ipython_demo.utils import generate_good_names
from .models import CallOptions, CallResponse, ShellInfo from .models import AppInfo, CallOptions, CallResponse, ShellInfo
from .shell import Shell from .shell import Shell
@@ -15,6 +17,10 @@ class App:
"""Coordinate named shells and expose frontend-shaped call responses.""" """Coordinate named shells and expose frontend-shaped call responses."""
def __init__(self): def __init__(self):
self.app_id = uuid4().hex
self.created_at = datetime.now(UTC).isoformat()
self._started_at = monotonic()
self._total_calls = 0
self.shells: dict[str, Shell] = {} self.shells: dict[str, Shell] = {}
self.last_shell: str | None = None self.last_shell: str | None = None
@@ -62,6 +68,17 @@ class App:
raise KeyError("No shell exists yet") raise KeyError("No shell exists yet")
return self.get_shell(self.last_shell) return self.get_shell(self.last_shell)
def info(self) -> AppInfo:
"""Return runtime metadata without initializing or inspecting shells."""
return AppInfo(
app_id=self.app_id,
created_at=self.created_at,
uptime_seconds=monotonic() - self._started_at,
shell_count=len(self.shells),
last_shell=self.last_shell,
total_calls=self._total_calls,
)
def get_shell_2(self, shell_name: Literal["last"] | str) -> ShellInfo: def get_shell_2(self, shell_name: Literal["last"] | str) -> ShellInfo:
"""Return frontend-safe metadata for a single shell.""" """Return frontend-safe metadata for a single shell."""
if shell_name == "last": if shell_name == "last":
@@ -80,6 +97,7 @@ class App:
options = CallOptions() options = CallOptions()
selected_name, shell = self._select_shell(shell_name) selected_name, shell = self._select_shell(shell_name)
call_id = uuid4().hex call_id = uuid4().hex
self._total_calls += 1
execution, events = shell.run_cell(code, call_id=call_id) execution, events = shell.run_cell(code, call_id=call_id)
return CallResponse( return CallResponse(
call_id=call_id, call_id=call_id,
+12
View File
@@ -116,3 +116,15 @@ class ShellInfo:
created_at: str created_at: str
last_used_at: str | None last_used_at: str | None
capabilities: tuple[str, ...] capabilities: tuple[str, ...]
@dataclass
class AppInfo:
"""Small, transport-safe summary of one running application instance."""
app_id: str
created_at: str
uptime_seconds: float
shell_count: int
last_shell: str | None
total_calls: int
+28 -9
View File
@@ -1,20 +1,32 @@
# FastMCP is not FastAPI: this module exposes MCP tools, not HTTP routes. # FastMCP is not FastAPI: this module exposes MCP tools, not HTTP routes.
from typing import Literal from typing import Annotated, Literal
from fastmcp import FastMCP from fastmcp import FastMCP
from pydantic import Field
import ipython_demo import ipython_demo
from ipython_demo.models import CallOptions, CallResponse, ShellInfo from ipython_demo.models import AppInfo, CallOptions, CallResponse, ShellInfo
app = FastMCP(name="IPython Demo") app = FastMCP(name="IPython Demo")
shell_app = ipython_demo.App() shell_app = ipython_demo.App()
@app.tool @app.tool(title="Get app info")
def get_app_info() -> AppInfo:
"""Return metadata about this running IPython application."""
return shell_app.info()
@app.tool(title="Run code")
def run_code( def run_code(
code: str, code: str,
shell_name: Literal["last", "new"] | str = "last", shell_name: Annotated[
Literal["last", "new"] | str,
Field(
description="The name of the shell to run the code in. Use 'new' to create a new shell, or 'last' to use the most recently selected shell."
),
] = "last",
options: CallOptions | None = None, options: CallOptions | None = None,
) -> CallResponse: ) -> CallResponse:
"""Run code in a named shell and return a frontend-facing call response. """Run code in a named shell and return a frontend-facing call response.
@@ -24,14 +36,21 @@ def run_code(
return shell_app.run_code(code, shell_name, options=options) return shell_app.run_code(code, shell_name, options=options)
@app.tool @app.tool(title="List shells")
def list_shells() -> list[ShellInfo]: def list_shells() -> list[ShellInfo]:
"""Return frontend-safe metadata for all known shells.""" """Return frontend-safe metadata for all known shells."""
return shell_app.list_shells() return shell_app.list_shells()
@app.tool @app.tool(title="Get shell")
def get_shell(shell_name: Literal["last"] | str) -> ShellInfo: def get_shell(
shell_name: Annotated[
Literal["last"] | str,
Field(
description="The name of the shell to retrieve. Use 'last' to get the most recently selected shell."
),
],
) -> ShellInfo:
"""Return frontend-safe metadata for a single shell.""" """Return frontend-safe metadata for a single shell."""
return ( return (
shell_app.get_last_shell() shell_app.get_last_shell()
@@ -40,13 +59,13 @@ def get_shell(shell_name: Literal["last"] | str) -> ShellInfo:
) )
@app.tool @app.tool(title="Create shell")
def create_shell() -> ShellInfo: def create_shell() -> ShellInfo:
"""Create a fresh named shell and return its metadata.""" """Create a fresh named shell and return its metadata."""
return shell_app.create_shell() return shell_app.create_shell()
@app.tool @app.tool(title="Get last shell")
def get_last_shell() -> ShellInfo: def get_last_shell() -> ShellInfo:
""" """
Return metadata for the most recently selected shell. Return metadata for the most recently selected shell.
+5
View File
@@ -10,6 +10,11 @@ app = FastAPI(title="IPython Demo", version="0.1.0")
ShellApp = App() # global, scary ShellApp = App() # global, scary
@app.get("/info")
def get_info():
return ShellApp.info()
@app.get("/shells") @app.get("/shells")
def list_shells(): def list_shells():
shells = ShellApp.list_shells() shells = ShellApp.list_shells()
+9 -1
View File
@@ -10,6 +10,7 @@ from pydantic import SecretStr
from .ipython_wrapper import ( from .ipython_wrapper import (
create_shell, create_shell,
get_app_info,
get_last_shell, get_last_shell,
get_shell, get_shell,
list_shells, list_shells,
@@ -38,7 +39,14 @@ llm = ChatOpenAI(
agent = create_agent( agent = create_agent(
llm, llm,
tools=[create_shell, get_last_shell, get_shell, list_shells, run_code], tools=[
get_app_info,
create_shell,
get_last_shell,
get_shell,
list_shells,
run_code,
],
checkpointer=None, # soon checkpointer=None, # soon
) )
+7 -1
View File
@@ -3,11 +3,17 @@ from typing import Literal
from langchain.tools import tool from langchain.tools import tool
from ipython_demo import App from ipython_demo import App
from ipython_demo.models import CallOptions, CallResponse, ShellInfo from ipython_demo.models import AppInfo, CallOptions, CallResponse, ShellInfo
app = App() # global, again app = App() # global, again
@tool
def get_app_info() -> AppInfo:
"""Return metadata about this running IPython application."""
return app.info()
# should be similar to mcp/webapp. # should be similar to mcp/webapp.
@tool @tool
def run_code( def run_code(
+48
View File
@@ -0,0 +1,48 @@
import unittest
from fastapi.testclient import TestClient
from ipython_demo import App, AppInfo
from ipython_mcp.app import app as mcp_app
from ipython_webapp.app import app as web_app
from langchain_demo.ipython_wrapper import get_app_info
class AppInfoTests(unittest.IsolatedAsyncioTestCase):
def test_app_info_describes_runtime_and_updates_after_execution(self):
app = App()
before = app.info()
self.assertIsInstance(before, AppInfo)
self.assertEqual(before.shell_count, 0)
self.assertIsNone(before.last_shell)
self.assertEqual(before.total_calls, 0)
self.assertGreaterEqual(before.uptime_seconds, 0)
app.run_code("1 + 1", shell_name="info-test")
after = app.info()
self.assertEqual(after.shell_count, 1)
self.assertEqual(after.last_shell, "info-test")
self.assertEqual(after.total_calls, 1)
self.assertGreaterEqual(after.uptime_seconds, before.uptime_seconds)
def test_web_app_exposes_info(self):
response = TestClient(web_app).get("/info")
self.assertEqual(response.status_code, 200)
self.assertIsInstance(response.json()["shell_count"], int)
async def test_mcp_app_exposes_info_tool(self):
tools = await mcp_app.list_tools()
self.assertIn("get_app_info", {tool.name for tool in tools})
def test_langchain_exposes_info_tool(self):
result = get_app_info.invoke({})
self.assertIsInstance(result, AppInfo)
if __name__ == "__main__":
unittest.main()
+1
View File
@@ -14,6 +14,7 @@ class MCPAppTests(unittest.IsolatedAsyncioTestCase):
self.assertEqual( self.assertEqual(
{tool.name for tool in tools}, {tool.name for tool in tools},
{ {
"get_app_info",
"run_code", "run_code",
"list_shells", "list_shells",
"get_shell", "get_shell",