Model Context Protocol Basics

python -m venv .venv
source .venv/bin/activate
pip install mcp  # official Python SDK; verify version at build

Host (your script)
  └─ Client session
       └─ Server (e.g. filesystem via npx)

# Example only: official filesystem server scoped to a demo folder
mkdir -p ./mcp-demo && echo "hello from MCP" > ./mcp-demo/note.txt
npx -y @modelcontextprotocol/server-filesystem "$(pwd)/mcp-demo"

import asyncio
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
 
params = StdioServerParameters(
    command="npx",
    args=["-y", "@modelcontextprotocol/server-filesystem", "./mcp-demo"],
)
 
async def main() -> None:
    async with stdio_client(params) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()
            print("MCP session initialized")
 
asyncio.run(main())

async def list_tools(session: ClientSession) -> None:
    result = await session.list_tools()
    for tool in result.tools:
        print(tool.name, "-", (tool.description or "")[:80])

async def call_list_dir(session: ClientSession) -> None:
    # Tool names vary by server version - list first, then call
    tools = await session.list_tools()
    names = {t.name for t in tools.tools}
    print("available:", sorted(names))
 
    # Common filesystem-style tool name patterns (verify against list_tools)
    name = next(n for n in names if "list" in n.lower() or "directory" in n.lower())
    result = await session.call_tool(name, arguments={"path": "."})
    print(result)

def mcp_tools_to_openai_shape(tools) -> list[dict]:
    out = []
    for t in tools:
        out.append(
            {
                "type": "function",
                "function": {
                    "name": t.name,
                    "description": t.description or t.name,
                    "parameters": t.inputSchema
                    or {"type": "object", "properties": {}},
                },
            }
        )
    return out

async def run_mcp_tool(session: ClientSession, name: str, args: dict) -> dict:
    allowed = {t.name for t in (await session.list_tools()).tools}
    if name not in allowed:
        return {"ok": False, "error": f"unknown_or_disallowed:{name}"}
    try:
        result = await session.call_tool(name, arguments=args)
        return {"ok": True, "content": str(result)}
    except Exception as exc:  # narrow in production
        return {"ok": False, "error": type(exc).__name__, "message": str(exc)[:300]}

import json
 
async def toy_loop(session: ClientSession, user: str, max_turns: int = 4) -> str:
    tools = (await session.list_tools()).tools
    schemas = mcp_tools_to_openai_shape(tools)
    messages = [
        {"role": "system", "content": "Use MCP tools for file facts. Be concise."},
        {"role": "user", "content": user},
    ]
    # Replace fake_decide with a real chat.completions call that has tools=schemas
    for _ in range(max_turns):
        decision = fake_decide(messages, schemas)  # your model adapter
        if decision["type"] == "final":
            return decision["content"]
        obs = await run_mcp_tool(session, decision["name"], decision["args"])
        messages.append({"role": "assistant", "content": json.dumps(decision)})
        messages.append({"role": "tool", "content": json.dumps(obs)})
    return "Stopped: max turns reached."

import asyncio
from agents import Agent, Runner
from agents.mcp import MCPServerStdio
 
async def main() -> None:
    async with MCPServerStdio(
        name="fs",
        params={
            "command": "npx",
            "args": ["-y", "@modelcontextprotocol/server-filesystem", "./mcp-demo"],
        },
    ) as server:
        agent = Agent(
            name="File assistant",
            instructions="Use MCP filesystem tools. Never invent file contents.",
            mcp_servers=[server],
        )
        result = await Runner.run(agent, "List files you can see.")
        print(result.final_output)
 
asyncio.run(main())

# Pattern sketch - verify package imports for your SDK version
from mcp.client.streamable_http import streamablehttp_client
from mcp import ClientSession
 
async def remote_main(url: str) -> None:
    async with streamablehttp_client(url) as (read, write, _get_session_id):
        async with ClientSession(read, write) as session:
            await session.initialize()
            tools = await session.list_tools()
            print([t.name for t in tools.tools])