MCP Adapters

import sys

from mcp.server.fastmcp import FastMCP

CORPUS = {
    "chunking": "Split documents into overlapping chunks before embedding them.",
    "pgvector": "pgvector stores embeddings in a vector column and supports HNSW indexes.",
    "tool-design": "Design narrow, typed tools and return actionable errors.",
}

mcp = FastMCP("docs-assistant", host="127.0.0.1", port=8765)


@mcp.tool()
def search_docs(query: str, top_k: int = 2) -> list[dict]:
    """Search the DocsAssistant corpus and return matching pages."""
    words = query.lower().split()
    hits = [
        {"page": page, "snippet": text}
        for page, text in CORPUS.items()
        if any(w in page or w in text.lower() for w in words)
    ]
    return hits[:top_k]


@mcp.tool()
def open_issue(title: str) -> str:
    """Open a GitHub issue (simulated: nothing leaves this process)."""
    if not title.strip():
        raise ValueError("title must not be empty")
    return f"created issue: {title}"


@mcp.resource("docs://index")
def page_index() -> str:
    """The list of corpus pages."""
    return "\n".join(CORPUS)


@mcp.resource("docs://pages/{page}")
def read_page(page: str) -> str:
    """Full text of one corpus page."""
    return CORPUS.get(page, "not found")


@mcp.prompt()
def summarize_page(page: str) -> str:
    """Prompt template that asks for a summary of a corpus page."""
    return f"Summarize the page '{page}' in two sentences for a new contributor."


if __name__ == "__main__":
    mcp.run(transport="streamable-http" if "--http" in sys.argv else "stdio")

import asyncio
import sys

from langchain_mcp_adapters.client import MultiServerMCPClient


async def main() -> None:
    client = MultiServerMCPClient(
        {
            "docs": {"transport": "stdio", "command": sys.executable, "args": ["docs_server.py"]},
            "docs_http": {"transport": "streamable_http", "url": "http://127.0.0.1:8765/mcp"},
        },
        tool_name_prefix=True,
    )
    tools = await client.get_tools()
    print(sorted(t.name for t in tools))
    search = next(t for t in tools if t.name == "docs_http_search_docs")
    print(search.args_schema["properties"])
    print(await search.ainvoke({"query": "chunking"}))
    print(sorted(t.name for t in await client.get_tools(server_name="docs")))


asyncio.run(main())

['docs_http_open_issue', 'docs_http_search_docs', 'docs_open_issue', 'docs_search_docs']
{'query': {'title': 'Query', 'type': 'string'}, 'top_k': {'default': 2, 'title': 'Top K', 'type': 'integer'}}
[{'type': 'text', 'text': '{\n  "page": "chunking",\n  "snippet": "Split documents into overlapping chunks before embedding them."\n}', 'id': 'lc_3aea1089-d04e-4dd1-9061-7b1d00c43aad'}]
['docs_open_issue', 'docs_search_docs']

import asyncio
import sys

from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain_mcp_adapters.tools import load_mcp_tools


async def main() -> None:
    client = MultiServerMCPClient(
        {"docs": {"transport": "stdio", "command": sys.executable, "args": ["docs_server.py"]}}
    )
    async with client.session("docs") as session:  # one subprocess for the whole block
        tools = await load_mcp_tools(session)
        search = next(t for t in tools if t.name == "search_docs")
        print(await search.ainvoke({"query": "pgvector", "top_k": 1}))
        print(await search.ainvoke({"query": "tool", "top_k": 1}))


asyncio.run(main())

import asyncio

from langchain.agents import create_agent
from langchain.agents.middleware import HumanInTheLoopMiddleware
from langchain_core.language_models.fake_chat_models import GenericFakeChatModel
from langchain_core.messages import AIMessage
from langchain_mcp_adapters.client import MultiServerMCPClient
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import Command


class ScriptedChat(GenericFakeChatModel):
    def bind_tools(self, tools, **kwargs):
        return self


def call(name, args, id_):
    return AIMessage(content="", tool_calls=[{"name": name, "args": args, "id": id_, "type": "tool_call"}])


async def main() -> None:
    client = MultiServerMCPClient(
        {"docs": {"transport": "streamable_http", "url": "http://127.0.0.1:8765/mcp"}}
    )
    tools = await client.get_tools()
    model = ScriptedChat(messages=iter([
        call("search_docs", {"query": "chunking"}, "c1"),
        call("open_issue", {"title": "Explain overlap in chunking"}, "c2"),
        AIMessage(content="I opened an issue about chunk overlap."),
    ]))
    agent = create_agent(
        model,
        tools,
        middleware=[HumanInTheLoopMiddleware(interrupt_on={"open_issue": True, "search_docs": False})],
        checkpointer=InMemorySaver(),
    )
    cfg = {"configurable": {"thread_id": "t1"}}
    out = await agent.ainvoke({"messages": [{"role": "user", "content": "Is chunking documented?"}]}, cfg)
    req = out["__interrupt__"][0].value["action_requests"][0]
    print("paused on:", req["name"], req["args"])
    out = await agent.ainvoke(Command(resume={"decisions": [{"type": "approve"}]}), cfg)
    for m in out["messages"]:
        print(type(m).__name__, "|", getattr(m, "name", None), "|", str(m.content)[:90].replace("\n", " "))


asyncio.run(main())

paused on: open_issue {'title': 'Explain overlap in chunking'}
HumanMessage | None | Is chunking documented?
AIMessage | None |
ToolMessage | search_docs | [{'type': 'text', 'text': '{\n  "page": "chunking",\n  "snippet": "Split documents into ov
AIMessage | None |
ToolMessage | open_issue | [{'type': 'text', 'text': 'created issue: Explain overlap in chunking', 'id': 'lc_e682571b
AIMessage | None | I opened an issue about chunk overlap.

import asyncio

from langchain_mcp_adapters.client import MultiServerMCPClient


async def main() -> None:
    client = MultiServerMCPClient(
        {"docs": {"transport": "streamable_http", "url": "http://127.0.0.1:8765/mcp"}}
    )
    blobs = await client.get_resources("docs", uris=["docs://index", "docs://pages/pgvector"])
    for b in blobs:
        print(b.metadata["uri"], "|", b.as_string()[:60].replace("\n", ", "))
    print(len(await client.get_resources("docs")), "resources listed")

    messages = await client.get_prompt("docs", "summarize_page", arguments={"page": "chunking"})
    print(type(messages[0]).__name__, messages[0].content)

    async with client.session("docs") as session:
        listed = await session.list_prompts()
        print([p.name for p in listed.prompts])
        tpl = await session.list_resource_templates()
        print([t.uriTemplate for t in tpl.resourceTemplates])


asyncio.run(main())

docs://index | chunking, pgvector, tool-design
docs://pages/pgvector | pgvector stores embeddings in a vector column and supports HNSW indexes.
1 resources listed
HumanMessage Summarize the page 'chunking' in two sentences for a new contributor.
['summarize_page']
['docs://pages/{page}']

import asyncio
from pathlib import Path

from langchain.mcp import MCPAdapter


async def main() -> None:
    config = {
        "mcpServers": {
            "local": {"command": "python", "args": ["fm_server.py"]},
            "remote": {"url": "http://127.0.0.1:8765/mcp"},
        }
    }
    async with MCPAdapter(config) as adapter:
        tools = await adapter.list_tools()
        print(sorted(t.name for t in tools))
    async with MCPAdapter(Path("fm_server.py")) as adapter:
        print([t.name for t in await adapter.list_tools()])
    async with MCPAdapter("http://127.0.0.1:8765/mcp") as adapter:
        print([t.name for t in await adapter.list_tools()])


asyncio.run(main())

['local_search_docs', 'remote_open_issue', 'remote_search_docs']
['search_docs']
['search_docs', 'open_issue']