11. Build Your First MCP Server

from mcp.server import Server, NotificationOptions
from mcp.server.models import InitializationOptions
from mcp.server.stdio import stdio_server
from mcp.types import (
    Tool, Resource, Prompt,
    TextContent, ResourceContents,
    GetPromptResult, PromptMessage,
    PromptArgument
)
import json
from pathlib import Path

# Create the server instance
server = Server("doc-assistant")

# --- TOOLS ---

@server.list_tools()
async def list_tools() -> list[Tool]:
    return [
        Tool(
            name="search_docs",
            description="Search documentation for a query",
            inputSchema={
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "The search query"
                    },
                    "max_results": {
                        "type": "integer",
                        "description": "Maximum results (default: 5)",
                        "default": 5
                    }
                },
                "required": ["query"]
            }
        ),
        Tool(
            name="get_file_content",
            description="Read a file from the docs directory",
            inputSchema={
                "type": "object",
                "properties": {
                    "path": {
                        "type": "string",
                        "description": "File path relative to docs directory"
                    }
                },
                "required": ["path"]
            }
        ),
        Tool(
            name="list_files",
            description="List files in the docs directory",
            inputSchema={
                "type": "object",
                "properties": {
                    "directory": {
                        "type": "string",
                        "description": "Directory to list (default: root)",
                        "default": "."
                    }
                }
            }
        )
    ]

@server.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
    if name == "search_docs":
        query = arguments["query"]
        max_results = arguments.get("max_results", 5)
        # Simulate search - in production, use a real search engine
        results = [
            f"Result {i}: Document about {query}"
            for i in range(min(max_results, 3))
        ]
        return [TextContent(
            type="text",
            text=f"Search results for '{query}':\n" + "\n".join(results)
        )]

    elif name == "get_file_content":
        path = arguments["path"]
        try:
            content = Path(f"./docs/{path}").read_text()
            return [TextContent(type="text", text=content)]
        except FileNotFoundError:
            return [TextContent(type="text", text=f"File not found: {path}")]

    elif name == "list_files":
        directory = arguments.get("directory", ".")
        files = [str(f) for f in Path(f"./docs/{directory}").iterdir()]
        return [TextContent(
            type="text",
            text="Files:\n" + "\n".join(files)
        )]

    raise ValueError(f"Unknown tool: {name}")

# --- RESOURCES ---

@server.list_resources()
async def list_resources() -> list[Resource]:
    return [
        Resource(
            uri="docs://overview",
            name="Documentation Overview",
            description="High-level overview of the documentation",
            mimeType="text/markdown"
        ),
        Resource(
            uri="docs://getting-started",
            name="Getting Started Guide",
            description="How to get started with the project",
            mimeType="text/markdown"
        )
    ]

@server.read_resource()
async def read_resource(uri: str) -> list[ResourceContents]:
    # Map URIs to content
    content_map = {
        "docs://overview": "# Documentation Overview\n\nThis is the documentation for our project.",
        "docs://getting-started": "# Getting Started\n\nFollow these steps to get started..."
    }

    if uri in content_map:
        return [ResourceContents(
            uri=uri,
            mimeType="text/markdown",
            text=content_map[uri]
        )]

    raise ValueError(f"Resource not found: {uri}")

# --- PROMPTS ---

@server.list_prompts()
async def list_prompts() -> list[Prompt]:
    return [
        Prompt(
            name="explain_concept",
            description="Get an explanation of a concept at a specified level",
            arguments=[
                PromptArgument(
                    name="concept",
                    description="The concept to explain",
                    required=True
                ),
                PromptArgument(
                    name="level",
                    description="Explanation depth (basic, intermediate, advanced)",
                    required=False
                )
            ]
        )
    ]

@server.get_prompt()
async def get_prompt(name: str, arguments: dict) -> GetPromptResult:
    if name == "explain_concept":
        concept = arguments["concept"]
        level = arguments.get("level", "intermediate")

        level_instructions = {
            "basic": "Use simple language and analogies. Assume no prior knowledge.",
            "intermediate": "Assume the reader has basic understanding. Focus on depth.",
            "advanced": "Use technical language. Cover edge cases and advanced patterns."
        }

        instruction = level_instructions.get(level, level_instructions["intermediate"])

        return GetPromptResult(
            messages=[
                PromptMessage(
                    role="system",
                    content={
                        "type": "text",
                        "text": f"You are an expert explaining '{concept}'.\n{instruction}\n\nStructure your explanation:\n1. Core concept (definition)\n2. How it works\n3. Why it matters\n4. Example\n5. Key takeaways"
                    }
                )
            ]
        )

    raise ValueError(f"Unknown prompt: {name}")

# --- RUN THE SERVER ---

async def main():
    async with stdio_server() as (read, write):
        await server.run(
            read, write,
            InitializationOptions(
                server_name="doc-assistant",
                server_version="1.0.0",
                capabilities=server.get_capabilities(
                    notification_options=NotificationOptions(),
                    experimental_capabilities={}
                )
            )
        )

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

import { Server } from "@modelcontextprotocol/sdk/server/index.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import {
  ListToolsRequestSchema,
  CallToolRequestSchema,
  ListResourcesRequestSchema,
  ReadResourceRequestSchema,
  ListPromptsRequestSchema,
  GetPromptRequestSchema
} from "@modelcontextprotocol/sdk/types.js";

const server = new Server(
  { name: "doc-assistant", version: "1.0.0" },
  { capabilities: { tools: {}, resources: {}, prompts: {} } }
);

// --- TOOLS ---

server.setRequestHandler(ListToolsRequestSchema, async () => ({
  tools: [{
    name: "search_docs",
    description: "Search documentation for a query",
    inputSchema: {
      type: "object",
      properties: {
        query: { type: "string", description: "The search query" },
        max_results: {
          type: "integer",
          description: "Maximum results",
          default: 5
        }
      },
      required: ["query"]
    }
  }, {
    name: "get_file_content",
    description: "Read a file from the docs directory",
    inputSchema: {
      type: "object",
      properties: {
        path: { type: "string", description: "File path relative to docs" }
      },
      required: ["path"]
    }
  }]
}));

server.setRequestHandler(CallToolRequestSchema, async (request) => {
  const { name, arguments: args } = request.params;

  if (name === "search_docs") {
    const query = args.query;
    const results = [`Result 1: Document about ${query}`, `Result 2: More about ${query}`];
    return {
      content: [{ type: "text", text: `Search results:\n${results.join("\n")}` }]
    };
  }

  if (name === "get_file_content") {
    return {
      content: [{ type: "text", text: `Content of ${args.path}...` }]
    };
  }

  throw new Error(`Unknown tool: ${name}`);
});

// --- RESOURCES ---

server.setRequestHandler(ListResourcesRequestSchema, async () => ({
  resources: [{
    uri: "docs://overview",
    name: "Documentation Overview",
    description: "High-level overview",
    mimeType: "text/markdown"
  }]
}));

server.setRequestHandler(ReadResourceRequestSchema, async (request) => {
  const { uri } = request.params;
  return {
    contents: [{
      uri,
      mimeType: "text/markdown",
      text: "# Overview\n\nDocumentation content..."
    }]
  };
});

// --- PROMPTS ---

server.setRequestHandler(ListPromptsRequestSchema, async () => ({
  prompts: [{
    name: "explain_concept",
    description: "Explain a concept at a specified level",
    arguments: [
      { name: "concept", description: "The concept to explain", required: true },
      { name: "level", description: "Explanation depth", required: false }
    ]
  }]
}));

server.setRequestHandler(GetPromptRequestSchema, async (request) => {
  const { name, arguments: args } = request.params;

  if (name === "explain_concept") {
    return {
      messages: [{
        role: "system",
        content: {
          type: "text",
          text: `Explain '${args.concept}' at ${args.level || "intermediate"} level.`
        }
      }]
    };
  }

  throw new Error(`Unknown prompt: ${name}`);
});

// --- RUN ---

const transport = new StdioServerTransport();
await server.connect(transport);
console.error("Doc Assistant MCP Server running on stdio");