Compare commits

...

1 commit

Author SHA1 Message Date
Marvin Context Protocol
36f86c6743 Proxy remote server instructions in FastMCPProxy
🤖 Generated with Claude Code

Co-authored-by: Jeremiah Lowin <jlowin@users.noreply.github.com>
2026-03-19 14:25:29 +00:00
2 changed files with 78 additions and 5 deletions

View file

@ -10,7 +10,8 @@ from __future__ import annotations
import base64
import inspect
import time
from collections.abc import Awaitable, Callable, Sequence
from collections.abc import AsyncIterator, Awaitable, Callable, Sequence
from contextlib import asynccontextmanager
from typing import TYPE_CHECKING, Any, cast
from urllib.parse import quote
@ -720,8 +721,16 @@ class ProxyProvider(Provider):
"""
return []
# lifespan() uses default implementation (empty context manager)
# because client cleanup is handled per-request
# -------------------------------------------------------------------------
# Instructions
# -------------------------------------------------------------------------
async def _fetch_remote_instructions(self) -> str | None:
"""Fetch instructions from the remote server."""
client = await self._get_client()
async with client:
result = await client.initialize()
return result.instructions
# -----------------------------------------------------------------------------
@ -798,6 +807,10 @@ class FastMCPProxy(FastMCP):
This is a convenience wrapper that creates a FastMCP server with a
ProxyProvider. For more control, use FastMCP with add_provider(ProxyProvider(...)).
By default, the proxy fetches the remote server's instructions during
startup and uses them as its own. Explicitly passing ``instructions``
overrides this behavior.
Example:
```python
from fastmcp.server import create_proxy
@ -828,10 +841,25 @@ class FastMCPProxy(FastMCP):
Can be either a synchronous or asynchronous function.
**kwargs: Additional settings for the FastMCP server.
"""
self._explicit_instructions = "instructions" in kwargs
super().__init__(**kwargs)
self.client_factory = client_factory
provider: Provider = ProxyProvider(client_factory)
self.add_provider(provider)
self._proxy_provider: ProxyProvider = ProxyProvider(client_factory)
self.add_provider(self._proxy_provider)
@asynccontextmanager
async def _lifespan_manager(self) -> AsyncIterator[None]:
async with super()._lifespan_manager():
if not self._explicit_instructions and self.instructions is None:
try:
instructions = (
await self._proxy_provider._fetch_remote_instructions()
)
if instructions is not None:
self.instructions = instructions
except Exception:
logger.debug("Failed to fetch remote server instructions for proxy")
yield
# -----------------------------------------------------------------------------

View file

@ -245,6 +245,51 @@ async def test_proxy_with_async_client_factory():
assert client.transport.url == "http://example.com/mcp/"
class TestProxyInstructions:
async def test_proxy_inherits_remote_instructions(self):
"""Proxy should inherit instructions from the remote server."""
server = FastMCP("RemoteServer", instructions="Remote server instructions")
@server.tool
def hello() -> str:
return "hi"
proxy = create_proxy(server)
assert proxy.instructions is None # not set yet
async with Client(proxy) as client:
result = await client.initialize()
assert result.instructions == "Remote server instructions"
async def test_proxy_explicit_instructions_override_remote(self):
"""Explicitly set instructions should take precedence over remote."""
server = FastMCP("RemoteServer", instructions="Remote instructions")
@server.tool
def hello() -> str:
return "hi"
proxy = create_proxy(server, instructions="My proxy instructions")
async with Client(proxy) as client:
result = await client.initialize()
assert result.instructions == "My proxy instructions"
async def test_proxy_no_instructions_when_remote_has_none(self):
"""Proxy instructions should remain None if remote has none."""
server = FastMCP("RemoteServer")
@server.tool
def hello() -> str:
return "hi"
proxy = create_proxy(server)
async with Client(proxy) as client:
result = await client.initialize()
assert result.instructions is None
class TestTools:
async def test_get_tools(self, proxy_server):
tools = await proxy_server.list_tools()