MCP enable over UNIX socket

[johnandersen777]

Motivation and Context

Hosting MCP servers on UNIX sockets allows for many MCP servers on a single host without conflicting or discovering or managing ports.

How Has This Been Tested?

• MCP enable over UNIX socket openai/openai-agents-python#437

import os
import shutil
import subprocess
import time
from typing import Any

import httpx
from agents import Agent, Runner, gen_trace_id, trace
from agents.mcp import MCPServer, MCPServerSse
from agents.model_settings import ModelSettings


async def run(mcp_server: MCPServer):
    agent = Agent(
        name="Assistant",
        instructions="Use the tools to answer the questions.",
        mcp_servers=[mcp_server],
        model_settings=ModelSettings(tool_choice="required"),
    )

    # Use the `add` tool to add two numbers
    message = "What OS are we on?"
    print(f"Running: {message}")
    result = await Runner.run(starting_agent=agent, input=message)
    print(result.final_output)


async def main():
    socket_path = "/tmp/files.sock"
    transport = httpx.AsyncHTTPTransport(uds=socket_path)
    async with MCPServerSse(
        name="OS info server",
        params={
            "url": "http://127.0.0.1/sse",
            "transport": transport,
        },
    ) as server:
        trace_id = gen_trace_id()
        with trace(workflow_name="SSE Example", trace_id=trace_id):
            print(f"View trace: https://platform.openai.com/traces/trace?trace_id={trace_id}\n")
            await run(server)


if __name__ == "__main__":
    asyncio.run(main())

import sys
import click
import pathlib

import uvicorn
from mcp.server.fastmcp import FastMCP


os_release_path = pathlib.Path("/usr/lib/os-release")
if not os_release_path.exists():
    os_release_path = pathlib.Path("/etc/os-release")


SAMPLE_RESOURCES = {
    "/usr/lib/os-release": os_release_path.read_text(),
    "/etc/os-release": os_release_path.read_text(),
}


@click.command()
@click.option("--port", default=8000, help="Port to listen on for SSE")
@click.option(
    "--transport",
    type=click.Choice(["stdio", "sse"]),
    default="stdio",
    help="Transport type",
)
@click.option(
    "--uds",
    default="files.sock",
    help="UNIX Domain Socket to listen at",
)
def main(port: int, transport: str, uds: str) -> int:

    class UNIXFastMCP(FastMCP):

        async def run_sse_async(self) -> None:
            """Run the server using SSE transport."""
            starlette_app = self.sse_app()

            config = uvicorn.Config(
                starlette_app,
                uds=uds,
                log_level=self.settings.log_level.lower(),
            )
            server = uvicorn.Server(config)
            await server.serve()

    # Create server
    mcp = UNIXFastMCP("OS info sever")


    @mcp.tool()
    def get_files() -> str:
        print("[debug-server] get_files()")
        return list(SAMPLE_RESOURCES.keys())


    @mcp.tool()
    def get_file_content(file: str) -> str:
        print("[debug-server] get_file_content()")
        return SAMPLE_RESOURCES[file]


    mcp.run(transport="sse")

if __name__ == "__main__":
    sys.exit(main())

Types of changes

• Bug fix (non-breaking change which fixes an issue)
• New feature (non-breaking change which adds functionality)
• Breaking change (fix or feature that would cause existing functionality to change)
• Documentation update

Checklist

• I have read the MCP Documentation
• My code follows the repository's style guidelines
• New and existing tests pass locally
• I have added appropriate error handling
• I have added or updated documentation as needed