MCP servers — extending AI applications

import json
import asyncio
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp import types
from openai import OpenAI

# ── Create the MCP server ──────────────────────────────────────────────

server = Server("ai-backend-tools")
openai_client = OpenAI()


# ── Define tools ───────────────────────────────────────────────────────

@server.list_tools()
async def list_tools() -> list[types.Tool]:
    """Tell clients what tools this server provides."""
    return [
        types.Tool(
            name="summarize_text",
            description="Summarize a piece of text in a given number of sentences.",
            inputSchema={
                "type": "object",
                "properties": {
                    "text":      {"type": "string", "description": "The text to summarize"},
                    "sentences": {"type": "integer", "description": "Number of sentences in the summary", "default": 3},
                },
                "required": ["text"],
            },
        ),
        types.Tool(
            name="count_words",
            description="Count the number of words in a text string.",
            inputSchema={
                "type": "object",
                "properties": {
                    "text": {"type": "string", "description": "The text to count words in"},
                },
                "required": ["text"],
            },
        ),
    ]


@server.call_tool()
async def call_tool(name: str, arguments: dict) -> list[types.TextContent]:
    """Execute a tool and return the result."""

    if name == "summarize_text":
        text      = arguments["text"]
        sentences = arguments.get("sentences", 3)
        response = openai_client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{
                "role": "user",
                "content": f"Summarize the following in exactly {sentences} sentences:\n\n{text}",
            }],
        )
        summary = response.choices[0].message.content
        return [types.TextContent(type="text", text=summary)]

    elif name == "count_words":
        count = len(arguments["text"].split())
        return [types.TextContent(type="text", text=str(count))]

    else:
        return [types.TextContent(type="text", text=f"Unknown tool: {name}")]


# ── Run the server ──────────────────────────────────────────────────────

async def main():
    async with stdio_server() as (read_stream, write_stream):
        await server.run(
            read_stream,
            write_stream,
            server.create_initialization_options(),
        )

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

@server.list_resources()
async def list_resources() -> list[types.Resource]:
    return [
        types.Resource(
            uri="conversations://recent",
            name="Recent Conversations",
            description="The 10 most recent conversations from the database",
            mimeType="application/json",
        ),
    ]

@server.read_resource()
async def read_resource(uri: str) -> str:
    if uri == "conversations://recent":
        # In practice: query your Postgres database here
        conversations = [
            {"id": 1, "title": "Python debugging session"},
            {"id": 2, "title": "API design review"},
        ]
        return json.dumps(conversations, indent=2)
    raise ValueError(f"Unknown resource: {uri}")

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

async def use_mcp_server():
    server_params = StdioServerParameters(
        command="python",
        args=["mcp_server.py"],
    )
    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()

            # List available tools
            tools = await session.list_tools()
            print("Available tools:", [t.name for t in tools.tools])

            # Call a tool
            result = await session.call_tool(
                "count_words",
                {"text": "The quick brown fox jumps over the lazy dog"},
            )
            print("Word count:", result.content[0].text)

asyncio.run(use_mcp_server())

from mcp.server.sse import SseServerTransport
from fastapi import FastAPI
import uvicorn

app = FastAPI()
transport = SseServerTransport("/messages/")

# Mount the MCP transport on your FastAPI app
app.mount("/mcp", transport.router)

# Register tools the same way as stdio
@server.list_tools()
async def list_tools() -> list[types.Tool]:
    ...  # same as before

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8001)