# server.py
from mcp.server.fastmcp import FastMCP

# Create an MCP server
mcp = FastMCP("Demo")


# Add an addition tool
@mcp.tool()
def add(a: int, b: int) -> int:
    """Add two numbers"""
    return a + b


# Add a dynamic greeting resource
@mcp.resource("greeting://{name}")
def get_greeting(name: str) -> str:
    """Get a personalized greeting"""
    return f"Hello, {name}!"

上面是stdio 通信方式
使用 mcp inspector 来查看 mcp的运行状态

输入上面token，选择stdio 通信，点击连接

右边点击list templates ,显示出所有的模板

资源


1. Resource Template

  • 通过 @mcp.resource("greeting://{name}") 装饰器定义

  • 包含动态路径参数 {name}（URL 中的占位符）

  • 作用：定义一类资源的生成规则

  • 实际调用时需要传入具体参数（如 name="Alice"）才能生成具体资源

  • 相当于一个工厂函数，根据输入参数动态生成资源

2. Resource

  • 静态的、可直接访问的端点（如 @mcp.resource("status") 定义 /status）

  • 无动态路径参数

  • 直接返回固定内容（或基于全局状态的动态内容）

修改代码添加 一个资源

@mcp.resource("docs://api-documentation")
def get_api_documentation() -> dict:
    """获取API文档"""
    return {
        "version": "1.0.0",
        "title": "演示API",
        "description": "这是一个MCP服务器的演示API",
        "endpoints": [
            {
                "path": "/greeting/{name}",
                "method": "GET",
                "description": "获取个性化问候语"
            },
            {
                "path": "/user/{id}",
                "method": "GET", 
                "description": "获取用户信息"
            }
        ],
        "contact": {
            "email": "[email protected]",
            "website": "https://example.com"
        }
    }

添加prompts

Prompts是MCP服务器提供的提示词生成器，它们：

  • 接收参数

  • 返回格式化的提示词

  • 帮助客户端生成高质量的AI提示词

# 添加代码审查提示词
@mcp.prompt("code_review")
def code_review_prompt(code: str) -> str:
    """生成代码审查提示词"""
    return f"""Please review the following code and provide feedback on:
1. Code quality and best practices
2. Potential bugs or issues
3. Performance considerations
4. Security concerns
5. Suggestions for improvement

Code to review:
{code}

Please be constructive and specific in your feedback."""


# 添加邮件写作提示词
@mcp.prompt("email_writer")
def email_writer_prompt(subject: str, recipient: str, purpose: str, tone: str = "professional") -> str:
    """生成邮件写作提示词"""
    return f"""Write a professional email with the following details:
- Subject: {subject}
- Recipient: {recipient}
- Purpose: {purpose}
- Tone: {tone}

Please make it clear, concise, and professional."""


# 添加数据分析提示词
@mcp.prompt("data_analysis")
def data_analysis_prompt(dataset_name: str, columns: str, analysis_task: str) -> str:
    """生成数据分析提示词"""
    return f"""Analyze the following dataset and provide insights:

Dataset: {dataset_name}
Columns: {columns}
Task: {analysis_task}

Please provide:
1. Summary statistics
2. Key findings
3. Visualizations suggestions
4. Recommendations"""


# 添加创意写作提示词
@mcp.prompt("creative_writing")
def creative_writing_prompt(genre: str, theme: str, characters: str, setting: str, length: str) -> str:
    """生成创意写作提示词"""
    return f"""Write a creative piece based on the following:
- Genre: {genre}
- Theme: {theme}
- Characters: {characters}
- Setting: {setting}
- Length: {length}

Please create an engaging and original story."""

Tool

使用客户端调用

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

# Create server parameters for stdio connection
server_params = StdioServerParameters(
    command="python",  # Executable
    args=["server.py"],  # Optional command line arguments
    env=None,  # Optional environment variables
)

async def run():
    async with stdio_client(server_params) as (read, write):
        async with ClientSession(
            read, write
        ) as session:
            # Initialize the connection
            await session.initialize()

            # List available resources
            print("=== LISTING RESOURCES ===")
            resources = await session.list_resources()
            for resource in resources:
                print(f"Resource: {resource}")

            # List available tools
            print("\n=== LISTING TOOLS ===")
            tools = await session.list_tools()
            for tool in tools.tools:
                print(f"Tool: {tool.name}")

            # List available prompts
            print("\n=== LISTING PROMPTS ===")
            prompts = await session.list_prompts()
            for prompt in prompts.prompts:
                print(f"Prompt: {prompt.name}")

            # Test greeting resource
            print("\n=== TESTING GREETING RESOURCE ===")
            try:
                content, mime_type = await session.read_resource("greeting://Alice")
                print(f"Greeting: {content}")
                print(f"MIME Type: {mime_type}")
            except Exception as e:
                print(f"Error reading greeting resource: {e}")

            # Test user resource
            print("\n=== TESTING USER RESOURCE ===")
            try:
                content, mime_type = await session.read_resource("user://1")
                print(f"User Info: {content}")
            except Exception as e:
                print(f"Error reading user resource: {e}")

            # Test API documentation resource
            print("\n=== TESTING API DOCUMENTATION RESOURCE ===")
            try:
                content, mime_type = await session.read_resource("docs://api-documentation")
                print(f"API Documentation: {content}")
            except Exception as e:
                print(f"Error reading API documentation: {e}")

            # Test add tool
            print("\n=== TESTING ADD TOOL ===")
            try:
                result = await session.call_tool("add", arguments={"a": 10, "b": 20})
                print(f"Add result: {result.content}")
            except Exception as e:
                print(f"Error calling add tool: {e}")

            # Test code review prompt
            print("\n=== TESTING CODE REVIEW PROMPT ===")
            try:
                prompt_result = await session.get_prompt("code_review", arguments={"code": "def hello():\n    print('Hello, World!')"})
                print(f"Prompt Result Object: {prompt_result}")
                print(f"Prompt Result Type: {type(prompt_result)}")
                print(f"Prompt Result Dir: {dir(prompt_result)}")
                # Try to access the prompt content
        
                print(f"Code Review Prompt (raw):\n{prompt_result}")
            except Exception as e:
                print(f"Error getting code review prompt: {e}")

            # Test email writer prompt
            print("\n=== TESTING EMAIL WRITER PROMPT ===")
            try:
                prompt_result = await session.get_prompt("email_writer", arguments={
                    "subject": "Meeting Request",
                    "recipient": "[email protected]",
                    "purpose": "Schedule a project review meeting",
                    "tone": "professional"
                })
                print(f"Email Writer Prompt Result: {prompt_result}")
                # Try to access the prompt content
             
                print(f"Email Writer Prompt (raw):\n{prompt_result}")
            except Exception as e:
                print(f"Error getting email writer prompt: {e}")

            # Test data analysis prompt
            print("\n=== TESTING DATA ANALYSIS PROMPT ===")
            try:
                prompt_result = await session.get_prompt("data_analysis", arguments={
                    "dataset_name": "sales_data_2024",
                    "columns": "date, revenue, region, product",
                    "analysis_task": "Identify top performing regions"
                })
                print(f"Data Analysis Prompt Result: {prompt_result}")
                # Try to access the prompt content

                print(f"Data Analysis Prompt (raw):\n{prompt_result}")
            except Exception as e:
                print(f"Error getting data analysis prompt: {e}")

            # Test creative writing prompt
            print("\n=== TESTING CREATIVE WRITING PROMPT ===")
            try:
                prompt_result = await session.get_prompt("creative_writing", arguments={
                    "genre": "science fiction",
                    "theme": "artificial intelligence",
                    "characters": "AI researcher, robot assistant",
                    "setting": "future laboratory",
                    "length": "short story"
                })
                print(f"Creative Writing Prompt Result: {prompt_result}")
                # Try to access the prompt content

                print(f"Creative Writing Prompt (raw):\n{prompt_result}")
            except Exception as e:
                print(f"Error getting creative writing prompt: {e}")

            print("\n=== TESTING COMPLETED ===")


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