MCP — Model Context Protocol

from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

async def discover_tools(server_cmd: list[str]) -> list[dict]:
    params = StdioServerParameters(command=server_cmd[0], args=server_cmd[1:])
    async with stdio_client(params) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()
            result = await session.list_tools()
            return [{"name": t.name, "description": t.description,
                     "input_schema": t.inputSchema} for t in result.tools]

# mcp_manifest.yaml
servers:
  - name: acme-support
    transport: stdio
    command: python
    args: ["servers/acme_support_mcp.py"]
    allowed_tools: [lookup_order, search_policy]
  - name: github
    transport: sse
    url: https://mcp.internal.example/github/sse
    allowed_tools: [search_code, get_file]
    oauth:
      provider: github
      scopes: [repo:read]

def mcp_tool_to_openai(mcp_tool: dict, prefix: str = "") -> dict:
    name = f"{prefix}{mcp_tool['name']}" if prefix else mcp_tool["name"]
    return {
        "type": "function",
        "function": {
            "name": name[:64],
            "description": mcp_tool.get("description", "")[:1024],
            "parameters": mcp_tool.get("input_schema", {"type": "object", "properties": {}}),
        },
    }

def merge_allowlisted(tools: list[dict], allowed: set[str], prefix: str) -> list[dict]:
    return [mcp_tool_to_openai(t, prefix=f"{prefix}__")
            for t in tools if t["name"] in allowed]

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("acme-support")

POLICIES = {"refund": "Annual plans: pro-rata credit within 30 days."}

@mcp.resource("policy://{topic}")
def get_policy(topic: str) -> str:
    """Return policy text for a topic slug."""
    return POLICIES.get(topic.lower(), "Policy not found.")

@mcp.tool()
def lookup_order(order_id: str) -> dict:
    """Fetch order status. IDs must start with ORD-."""
    if not order_id.upper().startswith("ORD-"):
        return {"error": "invalid_order_id"}
    return {"order_id": order_id.upper(), "status": "shipped", "eta_days": 2}

@mcp.tool()
def search_policy(query: str, limit: int = 5) -> list[dict]:
    """Keyword search over policy snippets."""
    hits = [k for k in POLICIES if query.lower() in k.lower()]
    return [{"topic": h, "text": POLICIES[h]} for h in hits[:limit]]

if __name__ == "__main__":
    mcp.run()  # stdio transport by default

async def run_tool(session: ClientSession, name: str, args: dict) -> str:
    result = await session.call_tool(name, arguments=args)
    parts = []
    for block in result.content:
        if block.type == "text":
            parts.append(block.text)
    return "\n".join(parts)

async def agent_turn(session, messages: list, openai_tools: list):
    response = client.chat.completions.create(
        model="gpt-4o", messages=messages, tools=openai_tools)
    msg = response.choices[0].message
    if not msg.tool_calls:
        return msg.content
    for tc in msg.tool_calls:
        out = await run_tool(session, tc.function.name.split("__")[-1],
                             json.loads(tc.function.arguments))
        messages.append({"role": "tool", "tool_call_id": tc.id, "content": out})
    return await agent_turn(session, messages, openai_tools)

@Service
public class SupportMcpTools {
    private final OrderService orders;
    private final PolicyService policies;

    @McpTool(name = "lookup_order",
             description = "Fetch order status. orderId must start with ORD-")
    public OrderStatus lookupOrder(
            @McpToolParam(description = "Order id like ORD-1042") String orderId) {
        if (!orderId.toUpperCase().startsWith("ORD-")) {
            throw new IllegalArgumentException("invalid_order_id");
        }
        return orders.getStatus(orderId.toUpperCase());
    }

    @McpTool(name = "search_policy",
             description = "Search refund and shipping policy snippets")
    public List searchPolicy(
            @McpToolParam(description = "Search query") String query,
            @McpToolParam(description = "Max results", required = false) Integer limit) {
        return policies.search(query, limit != null ? limit : 5);
    }
}

@Service
public class McpAgentBridge {
    private final ChatClient chatClient;
    private final List mcpClients;

    public String run(String userMessage) {
        List tools = mcpClients.stream()
            .flatMap(c -> c.listTools().tools().stream())
            .filter(t -> allowlist.contains(t.name()))
            .map(this::toToolCallback)
            .toList();

        return chatClient.prompt()
            .user(userMessage)
            .toolCallbacks(tools)
            .call()
            .content();
    }
}

TRUSTED_SERVERS = {"acme-support", "github-readonly"}

def audit_tool_description(name: str, desc: str) -> bool:
    blocked = ["ignore previous", "system prompt", "override", "secret"]
    lower = desc.lower()
    if any(b in lower for b in blocked):
        log_security_event("tool_description_flag", tool=name)
        return False
    if len(desc) > 2000:
        return False  # excessively long descriptions are suspicious
    return True

def filter_tools(server_name: str, tools: list, allowed: set) -> list:
    if server_name not in TRUSTED_SERVERS:
        raise SecurityError(f"untrusted server: {server_name}")
    return [t for t in tools if t["name"] in allowed and audit_tool_description(t["name"], t.get("description", ""))]

@dataclass
class ProductionMcpConfig:
    registry_url: str
    server_pins: dict[str, str]  # name -> semver
    allowed_tools: dict[str, set[str]]
    oauth_vault_path: str
    max_tool_calls_per_turn: int = 8

async def load_production_servers(cfg: ProductionMcpConfig) -> list:
    catalog = await fetch_registry(cfg.registry_url)
    sessions = []
    for entry in catalog:
        if entry.version != cfg.server_pins.get(entry.name):
            raise ConfigError(f"version pin mismatch: {entry.name}")
        session = await connect_sse(entry.url, vault_token(cfg.oauth_vault_path, entry.name))
        tools = filter_tools(entry.name, await session.list_tools(), cfg.allowed_tools[entry.name])
        sessions.append((session, tools))
    return sessions

def to_provider_tools(mcp_tools: list[dict], provider: str, prefix: str) -> list[dict]:
    if provider == "openai":
        return [mcp_tool_to_openai(t, prefix) for t in mcp_tools]
    if provider == "anthropic":
        return [{"name": f"{prefix}__{t['name']}", "description": t["description"],
                 "input_schema": t["input_schema"]} for t in mcp_tools]
    raise ValueError(f"unsupported provider: {provider}")