From c91b88f2172b0ca56460f5ba4ec864294b416f76 Mon Sep 17 00:00:00 2001 From: Eduard van Valkenburg Date: Fri, 12 Jun 2026 12:25:43 +0200 Subject: [PATCH] Python: add agent-framework-hosting-mcp channel (#6305) * feat(python): add agent-framework-hosting-mcp channel Add a hosting channel that exposes the host target (agent or workflow) as a single Model Context Protocol tool over Streamable HTTP. The tool invocation routes through the host pipeline (ChannelContext.run/ run_stream) so sessions, linking, and run/response hooks apply. Maps the MCP request context to a ChannelSession isolation key and ChannelIdentity, and forwards streaming output as MCP progress notifications. Includes tests, README, and workspace registration. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * Address MCP hosting channel review feedback Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --------- Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- python/packages/hosting-mcp/LICENSE | 21 + python/packages/hosting-mcp/README.md | 29 ++ .../agent_framework_hosting_mcp/__init__.py | 23 + .../agent_framework_hosting_mcp/_channel.py | 437 ++++++++++++++++++ python/packages/hosting-mcp/pyproject.toml | 102 ++++ .../tests/hosting_mcp/test_channel.py | 390 ++++++++++++++++ python/pyproject.toml | 2 + python/uv.lock | 23 + 8 files changed, 1027 insertions(+) create mode 100644 python/packages/hosting-mcp/LICENSE create mode 100644 python/packages/hosting-mcp/README.md create mode 100644 python/packages/hosting-mcp/agent_framework_hosting_mcp/__init__.py create mode 100644 python/packages/hosting-mcp/agent_framework_hosting_mcp/_channel.py create mode 100644 python/packages/hosting-mcp/pyproject.toml create mode 100644 python/packages/hosting-mcp/tests/hosting_mcp/test_channel.py diff --git a/python/packages/hosting-mcp/LICENSE b/python/packages/hosting-mcp/LICENSE new file mode 100644 index 0000000000..9e841e7a26 --- /dev/null +++ b/python/packages/hosting-mcp/LICENSE @@ -0,0 +1,21 @@ + MIT License + + Copyright (c) Microsoft Corporation. + + Permission is hereby granted, free of charge, to any person obtaining a copy + of this software and associated documentation files (the "Software"), to deal + in the Software without restriction, including without limitation the rights + to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + copies of the Software, and to permit persons to whom the Software is + furnished to do so, subject to the following conditions: + + The above copyright notice and this permission notice shall be included in all + copies or substantial portions of the Software. + + THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE + SOFTWARE diff --git a/python/packages/hosting-mcp/README.md b/python/packages/hosting-mcp/README.md new file mode 100644 index 0000000000..c6f01c4f45 --- /dev/null +++ b/python/packages/hosting-mcp/README.md @@ -0,0 +1,29 @@ +# agent-framework-hosting-mcp + +Model Context Protocol (MCP) tool channel for `agent-framework-hosting`. + +Exposes the hosted target (an `Agent` or a `Workflow`) as a single MCP tool over +the Streamable-HTTP transport, so MCP clients — other agents, IDE tooling — can +invoke it. Every call is routed through the host pipeline, so host sessions, +request metadata, and run/response hooks all apply. + +```python +from agent_framework.openai import OpenAIChatClient +from agent_framework_hosting import AgentFrameworkHost +from agent_framework_hosting_mcp import MCPChannel + +agent = OpenAIChatClient().as_agent(name="Assistant") + +host = AgentFrameworkHost(target=agent, channels=[MCPChannel()]) +host.serve(port=8000) +``` + +The Streamable-HTTP endpoint is mounted at `path` (default `/mcp`). The advertised +tool accepts `{"input": str, "session_id": str?}` and returns the target's reply +as MCP content blocks, including structured output when the agent returns one. +Pass `session_id` to continue a prior conversation (it maps onto the host +session). When `streaming=True` (default) incremental text is forwarded as MCP +progress notifications while the full reply is returned as the tool result. + +The base host plumbing lives in +[`agent-framework-hosting`](https://pypi.org/project/agent-framework-hosting/). diff --git a/python/packages/hosting-mcp/agent_framework_hosting_mcp/__init__.py b/python/packages/hosting-mcp/agent_framework_hosting_mcp/__init__.py new file mode 100644 index 0000000000..90abbd8776 --- /dev/null +++ b/python/packages/hosting-mcp/agent_framework_hosting_mcp/__init__.py @@ -0,0 +1,23 @@ +# Copyright (c) Microsoft. All rights reserved. + +"""Model Context Protocol (MCP) tool channel for :mod:`agent_framework_hosting`. + +Exposes the hosted target (an ``Agent`` or a ``Workflow``) as a single MCP +tool over the Streamable-HTTP transport so MCP clients — other agents, IDE +tooling — can invoke it. Routes through the host pipeline, so sessions, +request metadata, and hooks apply. +""" + +import importlib.metadata + +from ._channel import MCPChannel + +try: + __version__ = importlib.metadata.version(__name__) +except importlib.metadata.PackageNotFoundError: + __version__ = "0.0.0" + +__all__ = [ + "MCPChannel", + "__version__", +] diff --git a/python/packages/hosting-mcp/agent_framework_hosting_mcp/_channel.py b/python/packages/hosting-mcp/agent_framework_hosting_mcp/_channel.py new file mode 100644 index 0000000000..427268c4e1 --- /dev/null +++ b/python/packages/hosting-mcp/agent_framework_hosting_mcp/_channel.py @@ -0,0 +1,437 @@ +# Copyright (c) Microsoft. All rights reserved. + +"""``MCPChannel`` — exposes the hosted target as a Model Context Protocol tool. + +Mounts a Streamable-HTTP MCP endpoint that advertises a single tool. An MCP +client (another agent, an IDE, tooling) calls the tool with +``{"input": "...", "session_id": "..."}`` and receives the target's reply as +the tool result. + +Like the other ``agent-framework-hosting`` channels this routes through the +host pipeline (``ChannelContext.run`` / ``run_stream``) so session resolution, +request metadata, and run/response hooks all apply. The MCP ``tool/call`` +conversation key maps onto :class:`ChannelSession` (caller-supplied-session +family); the same single-tool shape works for an ``Agent`` or a ``Workflow`` +target (use a ``run_hook`` to reshape the free-form input into a workflow's +typed inputs). +""" + +from __future__ import annotations + +import base64 +import json +import re +from collections.abc import Mapping, Sequence +from contextlib import AbstractAsyncContextManager +from dataclasses import asdict, is_dataclass +from typing import Any, cast + +import mcp.types as types +from agent_framework import Content, Message +from agent_framework_hosting import ( + ChannelContext, + ChannelContribution, + ChannelIdentity, + ChannelRequest, + ChannelResponseHook, + ChannelRunHook, + ChannelSession, + HostedRunResult, + logger, +) +from mcp.server.lowlevel import Server +from mcp.server.streamable_http_manager import StreamableHTTPSessionManager +from pydantic import AnyUrl +from starlette.routing import Mount +from starlette.types import Receive, Scope, Send + +_DEFAULT_TOOL_NAME = "run_agent" +_DEFAULT_TOOL_DESCRIPTION = ( + "Invoke the hosted agent (or workflow) with a free-form text request and " + "return its reply. Pass an optional ``session_id`` to continue a prior " + "conversation." +) +_DATA_URI_PATTERN = re.compile(r"^data:(?P[^;]+);base64,(?P[A-Za-z0-9+/=]+)$") + + +def _mcp_uri(uri: str) -> AnyUrl: + """Build an MCP URI model from a string URI.""" + return AnyUrl(uri) + + +def _json_safe(value: Any) -> Any: + """Return a JSON-serializable representation for MCP structured content.""" + try: + return json.loads(json.dumps(value, default=str)) + except (TypeError, ValueError): + return str(value) + + +def _structured_content(value: Any) -> dict[str, Any] | None: + """Normalize an Agent Framework structured output value for MCP.""" + if value is None: + return None + + model_dump = getattr(value, "model_dump", None) + if callable(model_dump): + value = model_dump(mode="json") + elif is_dataclass(value) and not isinstance(value, type): + value = asdict(value) + + if isinstance(value, Mapping): + mapping_value = cast("Mapping[Any, Any]", value) # type: ignore[redundant-cast] + safe_value = _json_safe(dict(mapping_value)) + if isinstance(safe_value, dict): + safe_mapping = cast("Mapping[Any, Any]", safe_value) + return {str(key): item for key, item in safe_mapping.items()} + return {"value": safe_value} + safe_value = _json_safe(value) + return {"value": safe_value} + + +def _data_content_to_mcp(content: Content) -> list[types.ContentBlock]: + """Convert Agent Framework data content into the closest MCP content block.""" + if not content.uri: + return [] + match = _DATA_URI_PATTERN.match(content.uri) + if match is None: + logger.warning("MCPChannel could not parse data URI; omitted.") + return [] + + media_type = content.media_type or match.group("media_type") + data = match.group("data") + if media_type.startswith("image/"): + return [types.ImageContent(type="image", data=data, mimeType=media_type)] + if media_type.startswith("audio/"): + return [types.AudioContent(type="audio", data=data, mimeType=media_type)] + return [ + types.EmbeddedResource( + type="resource", + resource=types.BlobResourceContents(uri=_mcp_uri(content.uri), mimeType=media_type, blob=data), + ) + ] + + +def _content_to_mcp(content: Content) -> list[types.ContentBlock]: + """Convert one Agent Framework content item into MCP content blocks.""" + match content.type: + case "text": + return [types.TextContent(type="text", text=content.text or "")] + case "text_reasoning": + return [types.TextContent(type="text", text=content.text)] if content.text else [] + case "data": + return _data_content_to_mcp(content) + case "uri": + if not content.uri: + return [] + block: types.ContentBlock = types.ResourceLink( + type="resource_link", + name=content.uri, + uri=_mcp_uri(content.uri), + mimeType=content.media_type, + ) + return [block] + case "function_result": + if content.items: + blocks: list[types.ContentBlock] = [] + for item in content.items: + blocks.extend(_content_to_mcp(item)) + return blocks + return [types.TextContent(type="text", text=str(content.result or ""))] + case "error": + return [types.TextContent(type="text", text=content.message or content.error_details or "")] + case _: + logger.warning("MCPChannel does not support content type: %s. Omitted.", content.type) + return [] + + +def _value_to_mcp(value: Any) -> list[types.ContentBlock]: + """Convert a workflow output or fallback value into MCP content blocks.""" + if isinstance(value, Content): + return _content_to_mcp(value) + if isinstance(value, Message): + blocks: list[types.ContentBlock] = [] + for content in value.contents: + blocks.extend(_content_to_mcp(content)) + return blocks + if isinstance(value, str): + return [types.TextContent(type="text", text=value)] + if isinstance(value, bytes): + data = base64.b64encode(value).decode("utf-8") + return [ + types.EmbeddedResource( + type="resource", + resource=types.BlobResourceContents( + uri=_mcp_uri("data:application/octet-stream;base64," + data), + mimeType="application/octet-stream", + blob=data, + ), + ) + ] + return [types.TextContent(type="text", text=json.dumps(_json_safe(value), default=str))] + + +class MCPChannel: + """Exposes the hosted target as a single MCP tool over Streamable HTTP. + + Mounts the MCP Streamable-HTTP transport at ``path`` (default ``/mcp``). + The advertised tool accepts ``{"input": str, "session_id": str?}`` and + returns the target's reply as MCP content blocks. Agent structured outputs + are returned as MCP ``structuredContent``. + """ + + name = "mcp" + + def __init__( + self, + *, + path: str = "/mcp", + tool_name: str = _DEFAULT_TOOL_NAME, + tool_description: str = _DEFAULT_TOOL_DESCRIPTION, + server_name: str | None = None, + server_version: str | None = None, + streaming: bool = True, + json_response: bool = False, + stateless: bool = False, + run_hook: ChannelRunHook | None = None, + response_hook: ChannelResponseHook | None = None, + ) -> None: + """Create an MCP tool channel. + + Keyword Args: + path: Mount path for the Streamable-HTTP transport. Default ``/mcp``. + tool_name: Name of the advertised tool. Default ``run_agent``. + tool_description: Human-readable description advertised to clients. + server_name: MCP server name reported in the initialize handshake. + Defaults to the hosted target's ``name`` attribute when available. + server_version: Optional MCP server version string. + streaming: When ``True`` (default) the channel consumes the target + via :meth:`ChannelContext.run_stream` and forwards incremental + text to the client as MCP progress notifications (when the + client supplied a ``progressToken``). The full reply is always + returned as the tool result regardless of this flag. + json_response: Forwarded to :class:`StreamableHTTPSessionManager`. + When ``True`` the transport returns a single JSON response + instead of an SSE stream for each request. + stateless: Forwarded to :class:`StreamableHTTPSessionManager`. When + ``True`` the transport does not retain per-session state between + requests. + run_hook: Optional :data:`ChannelRunHook` invoked with the parsed + :class:`ChannelRequest` before the target runs. + response_hook: Optional :data:`ChannelResponseHook` invoked before + the channel serializes an originating reply into tool content. + """ + self.path = path + self.response_hook = response_hook + self._tool_name = tool_name + self._tool_description = tool_description + self._server_name = server_name + self._server_version = server_version + self._streaming = streaming + self._json_response = json_response + self._stateless = stateless + self._hook = run_hook + self._ctx: ChannelContext | None = None + self._server: Server[Any, Any] | None = None + self._session_manager: StreamableHTTPSessionManager | None = None + self._run_cm: AbstractAsyncContextManager[None] | None = None + + def contribute(self, context: ChannelContext) -> ChannelContribution: + """Capture the host context and mount the Streamable-HTTP transport.""" + self._ctx = context + self._server = self._build_server() + self._session_manager = StreamableHTTPSessionManager( + app=self._server, + json_response=self._json_response, + stateless=self._stateless, + ) + # StreamableHTTPSessionManager owns MCP initialize/session/progress semantics; + # mounting it keeps the channel on the real MCP HTTP transport. + return ChannelContribution( + routes=[Mount("/", app=self._handle_asgi)], + on_startup=[self._on_startup], + on_shutdown=[self._on_shutdown], + ) + + async def _handle_asgi(self, scope: Scope, receive: Receive, send: Send) -> None: + """ASGI entrypoint delegating to the MCP Streamable-HTTP session manager.""" + if self._session_manager is None: # pragma: no cover - guarded by lifecycle + raise RuntimeError("MCPChannel transport not initialized") + await self._session_manager.handle_request(scope, receive, send) + + async def _on_startup(self) -> None: + """Enter the session-manager task-group lifecycle on host startup.""" + if self._session_manager is None: # pragma: no cover - guarded by lifecycle + return + self._run_cm = self._session_manager.run() + await self._run_cm.__aenter__() + + async def _on_shutdown(self) -> None: + """Exit the session-manager task-group lifecycle on host shutdown.""" + if self._run_cm is not None: + await self._run_cm.__aexit__(None, None, None) + self._run_cm = None + + def _build_server(self) -> Server[Any, Any]: + """Build the low-level MCP server with the single host-routed tool.""" + target_name = getattr(self._ctx.target, "name", None) if self._ctx is not None else None + server_name = self._server_name or (target_name if isinstance(target_name, str) and target_name else None) + server: Server[Any, Any] = Server(name=server_name or "agent-framework-hosting", version=self._server_version) + tool = types.Tool( + name=self._tool_name, + description=self._tool_description, + inputSchema={ + "type": "object", + "properties": { + "input": { + "type": "string", + "description": "The request to send to the hosted agent or workflow.", + }, + "session_id": { + "type": "string", + "description": "Optional conversation id to continue a prior session.", + }, + }, + "required": ["input"], + }, + ) + + @server.list_tools() # type: ignore[no-untyped-call, untyped-decorator, misc] + async def _list_tools() -> list[types.Tool]: # noqa: RUF029 # pyright: ignore[reportUnusedFunction] + return [tool] + + @server.call_tool() # type: ignore[no-untyped-call, untyped-decorator, misc] + async def _call_tool(name: str, arguments: Mapping[str, Any]) -> types.CallToolResult: # pyright: ignore[reportUnusedFunction] + return await self._invoke_tool(arguments) + + return server + + async def _invoke_tool(self, arguments: Mapping[str, Any]) -> types.CallToolResult: + """Route a single ``tool/call`` through the host pipeline.""" + if self._ctx is None: # pragma: no cover - guarded by Channel lifecycle + raise RuntimeError("MCPChannel not initialized") + + text_input = arguments.get("input") + if not isinstance(text_input, str) or not text_input: + return types.CallToolResult( + content=[types.TextContent(type="text", text="Error: 'input' must be a non-empty string.")], + isError=True, + ) + session_id = arguments.get("session_id") + session = ChannelSession(isolation_key=session_id) if isinstance(session_id, str) and session_id else None + identity = ( + ChannelIdentity(channel=self.name, native_id=session_id) + if isinstance(session_id, str) and session_id + else None + ) + + channel_request = ChannelRequest( + channel=self.name, + operation="message.create", + input=text_input, + session=session, + stream=self._streaming, + identity=identity, + attributes={"tool_name": self._tool_name}, + ) + + if channel_request.stream: + result = await self._run_streaming(channel_request, protocol_request=dict(arguments)) + else: + result = await self._ctx.run( + channel_request, + run_hook=self._hook, + protocol_request=dict(arguments), + response_hook=self.response_hook, + channel_name=self.name, + ) + + return self._result_to_content(result) + + async def _run_streaming( + self, request: ChannelRequest, *, protocol_request: Mapping[str, Any] + ) -> HostedRunResult[Any]: + """Consume the target as a stream, forwarding progress, returning the full reply.""" + if self._ctx is None: # pragma: no cover - guarded by Channel lifecycle + raise RuntimeError("MCPChannel not initialized") + + progress_token, request_id = self._progress_context() + progress = 0.0 + stream = await self._ctx.run_stream( + request, + run_hook=self._hook, + protocol_request=protocol_request, + response_hook=self.response_hook, + channel_name=self.name, + ) + async for update in stream: + chunk = getattr(update, "text", None) + if not chunk: + continue + if progress_token is not None: + progress += 1.0 + try: + await self._send_progress(progress_token, progress, chunk, request_id) + except Exception: # pragma: no cover - progress is best-effort + logger.exception("MCPChannel progress notification failed") + return HostedRunResult(await stream.get_final_response()) + + def _progress_context(self) -> tuple[str | int | None, str | None]: + """Best-effort lookup of the active request's progress token + id.""" + if self._server is None: # pragma: no cover - guarded by lifecycle + return None, None + try: + ctx = self._server.request_context + except Exception: # pragma: no cover - no active request context + return None, None + token = ctx.meta.progressToken if ctx.meta is not None else None + request_id = str(ctx.request_id) + return token, request_id + + async def _send_progress( + self, + progress_token: str | int, + progress: float, + message: str, + request_id: str | None, + ) -> None: + """Send a single MCP progress notification for streamed text.""" + if self._server is None: # pragma: no cover - guarded by lifecycle + return + await self._server.request_context.session.send_progress_notification( + progress_token=progress_token, + progress=progress, + message=message, + related_request_id=request_id, + ) + + def _result_to_content(self, result: HostedRunResult[Any]) -> types.CallToolResult: + """Convert a host result into an MCP tool result.""" + response = result.result + content: list[types.ContentBlock] = [] + + messages = cast("Sequence[Any] | None", getattr(response, "messages", None)) + if messages: + for message in messages: + for item in cast("Sequence[Any]", getattr(message, "contents", None) or ()): + if isinstance(item, Content): + content.extend(_content_to_mcp(item)) + else: + content.append(types.TextContent(type="text", text=str(item))) + + get_outputs = getattr(response, "get_outputs", None) + if callable(get_outputs): + for output in cast("Sequence[Any]", get_outputs()): + content.extend(_value_to_mcp(output)) + + structured = _structured_content(getattr(response, "value", None)) + if not content: + text = getattr(response, "text", None) + if isinstance(text, str) and text: + content.append(types.TextContent(type="text", text=text)) + elif structured is not None: + content.append(types.TextContent(type="text", text=json.dumps(structured, indent=2))) + else: + content.append(types.TextContent(type="text", text="")) + + return types.CallToolResult(content=content, structuredContent=structured, isError=False) diff --git a/python/packages/hosting-mcp/pyproject.toml b/python/packages/hosting-mcp/pyproject.toml new file mode 100644 index 0000000000..39ac2ff27a --- /dev/null +++ b/python/packages/hosting-mcp/pyproject.toml @@ -0,0 +1,102 @@ +[project] +name = "agent-framework-hosting-mcp" +description = "Model Context Protocol (MCP) tool channel for agent-framework-hosting." +authors = [{ name = "Microsoft", email = "af-support@microsoft.com"}] +readme = "README.md" +requires-python = ">=3.10" +version = "1.0.0a260424" +license-files = ["LICENSE"] +urls.homepage = "https://aka.ms/agent-framework" +urls.source = "https://github.com/microsoft/agent-framework/tree/main/python" +urls.release_notes = "https://github.com/microsoft/agent-framework/releases?q=tag%3Apython-1&expanded=true" +urls.issues = "https://github.com/microsoft/agent-framework/issues" +classifiers = [ + "License :: OSI Approved :: MIT License", + "Development Status :: 3 - Alpha", + "Intended Audience :: Developers", + "Programming Language :: Python :: 3", + "Programming Language :: Python :: 3.10", + "Programming Language :: Python :: 3.11", + "Programming Language :: Python :: 3.12", + "Programming Language :: Python :: 3.13", + "Programming Language :: Python :: 3.14", + "Typing :: Typed", +] +dependencies = [ + "agent-framework-core>=1.2.0,<2", + "agent-framework-hosting>=1.0.0a260424,<2", + "mcp>=1.12,<2", + "starlette>=0.37", +] + +[tool.uv] +prerelease = "if-necessary-or-explicit" +environments = [ + "sys_platform == 'darwin'", + "sys_platform == 'linux'", + "sys_platform == 'win32'" +] + +[tool.uv-dynamic-versioning] +fallback-version = "0.0.0" + +[tool.pytest.ini_options] +testpaths = 'tests' +addopts = "-ra -q -r fEX" +asyncio_mode = "auto" +asyncio_default_fixture_loop_scope = "function" +filterwarnings = [] +timeout = 120 +markers = [ + "integration: marks tests as integration tests that require external services", +] + +[tool.ruff] +extend = "../../pyproject.toml" + +[tool.coverage.run] +omit = [ + "**/__init__.py" +] + +[tool.pyright] +extends = "../../pyproject.toml" +include = ["agent_framework_hosting_mcp"] +exclude = ['tests'] + +[tool.mypy] +plugins = ['pydantic.mypy'] +strict = true +python_version = "3.10" +ignore_missing_imports = true +disallow_untyped_defs = true +no_implicit_optional = true +check_untyped_defs = true +warn_return_any = true +show_error_codes = true +warn_unused_ignores = false +disallow_incomplete_defs = true +disallow_untyped_decorators = true + +[tool.bandit] +targets = ["agent_framework_hosting_mcp"] +exclude_dirs = ["tests"] + +[tool.poe] +executor.type = "uv" +include = "../../shared_tasks.toml" + +[tool.poe.tasks.mypy] +help = "Run MyPy for this package." +cmd = "mypy --config-file $POE_ROOT/pyproject.toml agent_framework_hosting_mcp" + +[tool.poe.tasks.test] +help = "Run the default unit test suite for this package." +cmd = 'pytest -m "not integration" --cov=agent_framework_hosting_mcp --cov-report=term-missing:skip-covered tests' + +[build-system] +requires = ["flit-core >= 3.11,<4.0"] +build-backend = "flit_core.buildapi" + +[dependency-groups] +dev = [] diff --git a/python/packages/hosting-mcp/tests/hosting_mcp/test_channel.py b/python/packages/hosting-mcp/tests/hosting_mcp/test_channel.py new file mode 100644 index 0000000000..f7874c0315 --- /dev/null +++ b/python/packages/hosting-mcp/tests/hosting_mcp/test_channel.py @@ -0,0 +1,390 @@ +# Copyright (c) Microsoft. All rights reserved. + +"""Unit tests for :class:`MCPChannel`.""" + +from __future__ import annotations + +import asyncio +from collections.abc import AsyncIterator, Awaitable, Sequence +from contextlib import asynccontextmanager +from dataclasses import dataclass, field +from typing import Any + +import mcp.types as types +import uvicorn +from agent_framework import AgentResponse, AgentResponseUpdate, Content, Message, ResponseStream +from agent_framework_hosting import AgentFrameworkHost, ChannelRequest, HostedRunResult +from mcp import ClientSession +from mcp.client.streamable_http import streamable_http_client +from mcp.shared.memory import create_connected_server_and_client_session +from starlette.types import ASGIApp + +from agent_framework_hosting_mcp import MCPChannel + +# --------------------------------------------------------------------------- # +# Fakes # +# --------------------------------------------------------------------------- # + + +@dataclass +class _FakeResp: + text: str + messages: list[Message] = field(default_factory=list) + value: Any | None = None + + +@dataclass +class _FakeUpdate: + text: str + contents: list[Content] = field(default_factory=list) + message_id: str | None = None + + +class _FakeStream: + def __init__(self, chunks: list[str], final: _FakeResp | None = None) -> None: + self._chunks = chunks + self._final = final or _FakeResp(text="".join(chunks)) + + def __aiter__(self) -> AsyncIterator[_FakeUpdate]: + async def _gen() -> AsyncIterator[_FakeUpdate]: + for c in self._chunks: + yield _FakeUpdate(text=c) + + return _gen() + + async def get_final_response(self) -> _FakeResp: + return self._final + + +@dataclass +class _FakeTarget: + name: str = "Assistant" + description: str = "A helpful assistant." + + +class _FakeContext: + """Minimal stand-in for :class:`ChannelContext`.""" + + def __init__( + self, + *, + reply: str = "hello", + chunks: list[str] | None = None, + contents: list[Content] | None = None, + structured: Any | None = None, + ) -> None: + self.target = _FakeTarget() + self._reply = reply + self._chunks = chunks or [reply] + self._contents = contents or [Content.from_text(text=reply)] + self._structured = structured + self.requests: list[ChannelRequest] = [] + + async def run( + self, + request: ChannelRequest, + *, + run_hook: Any | None = None, + protocol_request: Any | None = None, + response_hook: Any | None = None, + channel_name: str | None = None, + ) -> HostedRunResult[Any]: + if run_hook is not None: + maybe_request = run_hook(request, target=self.target, protocol_request=protocol_request) + if isinstance(maybe_request, Awaitable): + request = await maybe_request + else: + request = maybe_request + self.requests.append(request) + message = Message(role="assistant", contents=self._contents) + result = HostedRunResult(_FakeResp(text=self._reply, messages=[message], value=self._structured)) + if response_hook is not None: + maybe_result = response_hook(result, request=request, channel_name=channel_name or request.channel) + if isinstance(maybe_result, Awaitable): + return await maybe_result + return maybe_result + return result + + async def run_stream( + self, + request: ChannelRequest, + *, + run_hook: Any | None = None, + protocol_request: Any | None = None, + stream_update_hook: Any | None = None, + response_hook: Any | None = None, + channel_name: str | None = None, + ) -> _FakeStream: + if run_hook is not None: + maybe_request = run_hook(request, target=self.target, protocol_request=protocol_request) + if isinstance(maybe_request, Awaitable): + request = await maybe_request + else: + request = maybe_request + self.requests.append(request) + result = HostedRunResult(_FakeResp(text="".join(self._chunks), value=self._structured)) + if response_hook is not None: + maybe_result = response_hook(result, request=request, channel_name=channel_name or request.channel) + if isinstance(maybe_result, Awaitable): + result = await maybe_result + else: + result = maybe_result + return _FakeStream(self._chunks, final=result.result) + + +def _make_channel(ctx: _FakeContext, **kwargs: Any) -> MCPChannel: + channel = MCPChannel(**kwargs) + channel.contribute(ctx) # type: ignore[arg-type] + return channel + + +class _HostedAgent: + name = "HostedAssistant" + description = "A hosted test assistant." + + async def run(self, messages: Any = None, *, stream: bool = False, **_kwargs: Any) -> Any: + text = messages.text if isinstance(messages, Message) else str(messages) + if stream: + updates = [AgentResponseUpdate(contents=[Content.from_text(text=f"host: {text}")], role="assistant")] + + async def _gen() -> AsyncIterator[AgentResponseUpdate]: + for update in updates: + yield update + + async def _finalize(items: Sequence[AgentResponseUpdate]) -> AgentResponse: # noqa: RUF029 + return AgentResponse.from_updates(items) + + return ResponseStream[AgentResponseUpdate, AgentResponse](_gen(), finalizer=_finalize) + return AgentResponse(messages=[Message(role="assistant", contents=[Content.from_text(text=f"host: {text}")])]) + + +@asynccontextmanager +async def _serve_app(app: ASGIApp, *, port: int) -> AsyncIterator[str]: + config = uvicorn.Config(app, host="127.0.0.1", port=port, log_level="warning", lifespan="on") + server = uvicorn.Server(config) + task = asyncio.create_task(server.serve()) + try: + for _ in range(100): + if server.started: + break + await asyncio.sleep(0.01) + else: + raise RuntimeError("Test MCP server did not start") + yield f"http://127.0.0.1:{port}" + finally: + server.should_exit = True + await task + + +# --------------------------------------------------------------------------- # +# Tests # +# --------------------------------------------------------------------------- # + + +async def test_list_tools_advertises_single_configured_tool() -> None: + ctx = _FakeContext() + channel = _make_channel(ctx, tool_name="ask", tool_description="Ask the assistant.") + async with create_connected_server_and_client_session(channel._server) as client: # type: ignore[arg-type] + result = await client.list_tools() + assert len(result.tools) == 1 + tool = result.tools[0] + assert tool.name == "ask" + assert tool.description == "Ask the assistant." + assert tool.inputSchema["required"] == ["input"] + assert set(tool.inputSchema["properties"]) == {"input", "session_id"} + + +async def test_initialize_uses_target_name_by_default() -> None: + ctx = _FakeContext() + channel = _make_channel(ctx) + async with create_connected_server_and_client_session(channel._server) as client: # type: ignore[arg-type] + result = await client.initialize() + assert result.serverInfo.name == "Assistant" + + +async def test_call_tool_routes_through_host_and_returns_text() -> None: + ctx = _FakeContext(reply="hi back", chunks=["hi", " back"]) + channel = _make_channel(ctx, streaming=False) + async with create_connected_server_and_client_session(channel._server) as client: # type: ignore[arg-type] + result = await client.call_tool("run_agent", {"input": "hello", "session_id": "conv-1"}) + assert not result.isError + assert isinstance(result.content[0], types.TextContent) + assert result.content[0].text == "hi back" + # The channel built a channel-neutral request routed through the host. + assert len(ctx.requests) == 1 + request = ctx.requests[0] + assert request.channel == "mcp" + assert request.operation == "message.create" + assert request.input == "hello" + assert request.session is not None + assert request.session.isolation_key == "conv-1" + assert request.identity is not None + assert request.identity.native_id == "conv-1" + + +async def test_call_tool_returns_rich_content_and_structured_output() -> None: + ctx = _FakeContext( + contents=[ + Content.from_text(text="text"), + Content.from_data(data=b"image-bytes", media_type="image/png"), + Content.from_data(data=b"audio-bytes", media_type="audio/wav"), + Content.from_data(data=b"raw-bytes", media_type="application/octet-stream"), + Content.from_uri(uri="https://example.com/file.json", media_type="application/json"), + ], + structured={"answer": 42}, + ) + channel = _make_channel(ctx, streaming=False) + async with create_connected_server_and_client_session(channel._server) as client: # type: ignore[arg-type] + result = await client.call_tool("run_agent", {"input": "hello"}) + + assert result.structuredContent == {"answer": 42} + assert [item.type for item in result.content] == ["text", "image", "audio", "resource", "resource_link"] + assert result.content[0].text == "text" # type: ignore[union-attr] + + +async def test_call_tool_streaming_aggregates_chunks() -> None: + ctx = _FakeContext(chunks=["foo", "bar", "baz"]) + channel = _make_channel(ctx, streaming=True) + async with create_connected_server_and_client_session(channel._server) as client: # type: ignore[arg-type] + result = await client.call_tool("run_agent", {"input": "hello"}) + assert result.content[0].text == "foobarbaz" # type: ignore[union-attr] + # No session_id supplied -> no session / identity. + assert ctx.requests[0].session is None + assert ctx.requests[0].identity is None + + +async def test_call_tool_rejects_empty_input() -> None: + ctx = _FakeContext() + channel = _make_channel(ctx) + async with create_connected_server_and_client_session(channel._server) as client: # type: ignore[arg-type] + result = await client.call_tool("run_agent", {"input": ""}) + assert result.isError + assert "non-empty string" in result.content[0].text # type: ignore[union-attr] + assert ctx.requests == [] + + +async def test_run_hook_can_reshape_request() -> None: + ctx = _FakeContext(reply="ok") + + async def _hook(request: ChannelRequest, *, target: Any, protocol_request: Any) -> ChannelRequest: + import dataclasses + + return dataclasses.replace(request, attributes={**dict(request.attributes), "hooked": True}) + + channel = _make_channel(ctx, streaming=False, run_hook=_hook) + async with create_connected_server_and_client_session(channel._server) as client: # type: ignore[arg-type] + await client.call_tool("run_agent", {"input": "hello"}) + assert ctx.requests[0].attributes.get("hooked") is True + + +async def test_response_hook_can_shape_originating_reply() -> None: + ctx = _FakeContext(reply="original") + + async def _hook( + result: HostedRunResult[Any], + *, + request: ChannelRequest, + channel_name: str, + ) -> HostedRunResult[Any]: + assert channel_name == "mcp" + assert request.channel == "mcp" + return HostedRunResult(_FakeResp(text="hooked")) + + channel = _make_channel(ctx, streaming=False, response_hook=_hook) + async with create_connected_server_and_client_session(channel._server) as client: # type: ignore[arg-type] + result = await client.call_tool("run_agent", {"input": "hello"}) + assert result.content[0].text == "hooked" # type: ignore[union-attr] + + +async def test_streaming_response_hook_shapes_final_reply() -> None: + ctx = _FakeContext(chunks=["raw"]) + + async def _hook( + result: HostedRunResult[Any], + *, + request: ChannelRequest, + channel_name: str, + ) -> HostedRunResult[Any]: + return HostedRunResult(_FakeResp(text=f"{channel_name}:{request.channel}:{result.result.text}")) + + channel = _make_channel(ctx, streaming=True, response_hook=_hook) + async with create_connected_server_and_client_session(channel._server) as client: # type: ignore[arg-type] + result = await client.call_tool("run_agent", {"input": "hello"}) + assert result.content[0].text == "mcp:mcp:raw" # type: ignore[union-attr] + + +def test_default_path_and_name() -> None: + channel = MCPChannel() + assert channel.name == "mcp" + assert channel.path == "/mcp" + + +def test_content_conversion_handles_non_text_shapes() -> None: + from agent_framework_hosting_mcp._channel import _content_to_mcp, _structured_content, _value_to_mcp + + @dataclass + class StructuredValue: + answer: int + + circular: list[Any] = [] + circular.append(circular) + + assert _structured_content(None) is None + assert _structured_content(StructuredValue(answer=42)) == {"answer": 42} + assert _structured_content(circular) == {"value": "[[...]]"} + assert _content_to_mcp(Content("data", uri="not-a-data-uri", media_type="application/octet-stream")) == [] + assert _content_to_mcp(Content("text_reasoning", text="because"))[0].text == "because" # type: ignore[union-attr] + assert ( + _content_to_mcp(Content.from_function_result("call-1", result=[Content.from_text("nested")]))[0].text + == "nested" + ) # type: ignore[union-attr] + assert _content_to_mcp(Content.from_function_result("call-1", result={"x": 1}))[0].text == '{"x": 1}' # type: ignore[union-attr] + assert _content_to_mcp(Content.from_error(message="bad"))[0].text == "bad" # type: ignore[union-attr] + assert _content_to_mcp(Content.from_function_call("call-1", "tool")) == [] + assert _value_to_mcp(Message(role="assistant", contents=[Content.from_text("message")]))[0].text == "message" # type: ignore[union-attr] + assert _value_to_mcp(b"bytes")[0].type == "resource" + assert _value_to_mcp({"x": 1})[0].text == '{"x": 1}' # type: ignore[union-attr] + + +def test_result_conversion_handles_workflow_and_fallback_shapes() -> None: + class WorkflowResult: + value = None + + def get_outputs(self) -> list[Message]: + return [Message(role="assistant", contents=[Content.from_text("workflow")])] + + @dataclass + class TextOnlyResult: + text: str + value: Any | None = None + + channel = MCPChannel() + + workflow_result = channel._result_to_content(HostedRunResult(WorkflowResult())) + assert workflow_result.content[0].text == "workflow" # type: ignore[union-attr] + + text_result = channel._result_to_content(HostedRunResult(TextOnlyResult(text="fallback"))) + assert text_result.content[0].text == "fallback" # type: ignore[union-attr] + + structured_result = channel._result_to_content(HostedRunResult(TextOnlyResult(text="", value={"x": 1}))) + assert structured_result.structuredContent == {"x": 1} + assert structured_result.content[0].text == '{\n "x": 1\n}' # type: ignore[union-attr] + + empty_result = channel._result_to_content(HostedRunResult(TextOnlyResult(text=""))) + assert empty_result.content[0].text == "" # type: ignore[union-attr] + + +async def test_http_mcp_client_can_call_hosted_channel(unused_tcp_port: int) -> None: + host = AgentFrameworkHost(target=_HostedAgent(), channels=[MCPChannel(streaming=False)]) + + async with ( + _serve_app(host.app, port=unused_tcp_port) as base_url, + streamable_http_client(f"{base_url}/mcp/") as (read_stream, write_stream, _), + ClientSession(read_stream, write_stream) as session, + ): + await session.initialize() + tools = await session.list_tools() + result = await session.call_tool("run_agent", {"input": "hello", "session_id": "conv-1"}) + + assert [tool.name for tool in tools.tools] == ["run_agent"] + assert result.content[0].text == "host: hello" # type: ignore[union-attr] diff --git a/python/pyproject.toml b/python/pyproject.toml index ea9bcd3dd0..12d2a49154 100644 --- a/python/pyproject.toml +++ b/python/pyproject.toml @@ -91,6 +91,7 @@ agent-framework-hosting-telegram = { workspace = true } agent-framework-hosting-activity-protocol = { workspace = true } agent-framework-hosting-discord = { workspace = true } agent-framework-hosting-a2a = { workspace = true } +agent-framework-hosting-mcp = { workspace = true } agent-framework-hyperlight = { workspace = true } agent-framework-lab = { workspace = true } agent-framework-mem0 = { workspace = true } @@ -218,6 +219,7 @@ executionEnvironments = [ { root = "packages/hosting-telegram/tests", reportPrivateUsage = "none" }, { root = "packages/hosting-activity-protocol/tests", reportPrivateUsage = "none" }, { root = "packages/hosting-a2a/tests", reportPrivateUsage = "none" }, + { root = "packages/hosting-mcp/tests", reportPrivateUsage = "none" }, { root = "packages/lab/gaia/tests", reportPrivateUsage = "none" }, { root = "packages/lab/lightning/tests", reportPrivateUsage = "none" }, { root = "packages/lab/tau2/tests", reportPrivateUsage = "none" }, diff --git a/python/uv.lock b/python/uv.lock index 0a0becc285..17b3325aaf 100644 --- a/python/uv.lock +++ b/python/uv.lock @@ -52,6 +52,7 @@ members = [ "agent-framework-hosting-activity-protocol", "agent-framework-hosting-discord", "agent-framework-hosting-invocations", + "agent-framework-hosting-mcp", "agent-framework-hosting-responses", "agent-framework-hosting-telegram", "agent-framework-hyperlight", @@ -723,6 +724,28 @@ requires-dist = [ { name = "agent-framework-hosting", editable = "packages/hosting" }, ] +[[package]] +name = "agent-framework-hosting-mcp" +version = "1.0.0a260424" +source = { editable = "packages/hosting-mcp" } +dependencies = [ + { name = "agent-framework-core", marker = "sys_platform == 'darwin' or sys_platform == 'linux' or sys_platform == 'win32'" }, + { name = "agent-framework-hosting", marker = "sys_platform == 'darwin' or sys_platform == 'linux' or sys_platform == 'win32'" }, + { name = "mcp", extra = ["ws"], marker = "sys_platform == 'darwin' or sys_platform == 'linux' or sys_platform == 'win32'" }, + { name = "starlette", marker = "sys_platform == 'darwin' or sys_platform == 'linux' or sys_platform == 'win32'" }, +] + +[package.metadata] +requires-dist = [ + { name = "agent-framework-core", editable = "packages/core" }, + { name = "agent-framework-hosting", editable = "packages/hosting" }, + { name = "mcp", specifier = ">=1.12,<2" }, + { name = "starlette", specifier = ">=0.37" }, +] + +[package.metadata.requires-dev] +dev = [] + [[package]] name = "agent-framework-hosting-responses" version = "1.0.0a260424"