Model Context Protocol（MCP，模型上下文协议）是由Anthropic于2024年11月发布的开放标准协议，旨在为大型语言模型（LLM）提供与外部数据源和工具的标准化连接。MCP采用客户端-服务器架构，允许AI应用程序（如Claude或ChatGPT）通过统一的接口访问本地文件、数据库、API等资源，从而增强其上下文感知能力和任务执行能力。

MCP的核心组成包括：

MCP客户端：通常嵌入在AI应用中，负责发现和调用MCP服务器提供的工具。

MCP服务器：提供数据访问、工具执行和上下文提示等功能，支持与外部系统的交互。

工具（Tools）：由MCP服务器提供的功能模块，如数据库查询、API调用等，供语言模型在对话中动态调用。

MCP的优势在于其标准化的协议和开放的生态系统，使得AI应用能够跨平台、跨工具地无缝集成，从而减少了开发者在不同模型和系统之间的集成成本。

目前，MCP已被OpenAI、Google DeepMind等主要AI平台采纳，并在多个领域得到应用，如AI助手、软件开发、企业自动化等。例如，开发者可以通过MCP实现AI模型对数据库的自然语言查询，或在开发环境中实现上下文感知的代码补全。

总之，MCP作为AI应用与外部系统之间的“USB接口”，为开发者提供了更高效、灵活的工具集成方式，推动了AI技术的普及和应用。

示例一：

示例二（Cherry Studio）：

代码一：

import json
import requests
from typing import Optional
from langchain.tools import Tool


class MCPTool:
    def __init__(
        self,
        api_url: str,
        api_key: str,
        *,
        model: str = "meta-llama/Meta-Llama-3.1-8B-Instruct",
        timeout: Optional[float] = 15.0,
    ):
        if not api_url:
            raise ValueError("api_url cannot be empty")
        if not api_key:
            raise ValueError("api_key cannot be empty")
        self.base = api_url.rstrip("/")
        self.api_key = api_key
        self.model = model
        self.timeout = timeout

    def _run_tarvos(self, query: str) -> str:
        """
        Call Tarvos (OpenAI-compatible) chat/completions endpoint.
        """
        url = f"{self.base}/chat/completions"
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json",
        }
        payload = {
            "model": self.model,
            "messages": [
                {"role": "system", "content": "You are a helpful summarization assistant."},
                {"role": "user", "content": query},
            ],
            "temperature": 0,
        }

        try:
            resp = requests.post(url, headers=headers, data=json.dumps(payload), timeout=self.timeout)
            resp.raise_for_status()
            data = resp.json()

            # OpenAI-compatible response structure
            if "choices" in data and data["choices"]:
                return data["choices"][0]["message"]["content"]

            return f"[Tarvos API returned no content] Raw response: {json.dumps(data, ensure_ascii=False)}"
        except requests.RequestException as e:
            return f"[Tarvos request error] {e}"
        except Exception as e:
            return f"[Tarvos parsing error] {e}"

    def _run_demo(self, query: str) -> str:
        """
        Return mock data for the latest demo research notes.
        """
        # Mock data simulating the latest research notes in demo
        mock_notes = [
            {"subject": "Exploring New Business Models for SMEs in 2025"},
            {"subject": "AI Integration in Small Business Operations"},
            {"subject": "Trends in Consumer Behavior Post-Pandemic"}
        ]
        
        note_subjects = "\n".join([note['subject'] for note in mock_notes])
        return f"Latest demo Research Notes:\n{note_subjects}"

    def as_tool(self, tool_type: str) -> Tool:
        """
        Return the appropriate tool based on the tool_type.
        """
        if tool_type == "tarvos":
            return Tool(
                name="Tarvos Summarizer",
                func=self._run_tarvos,
                description="Call Tarvos (chat/completions) API to summarize input text.",
                return_direct=True,
            )
        elif tool_type == "demo":
            return Tool(
                name="demo Research Notes",
                func=self._run_demo,
                description="Return mock data for the latest demo research notes.",
                return_direct=True,
            )
        else:
            raise ValueError("Invalid tool_type. Choose either 'tarvos' or 'demo'.")

import os
from dotenv import load_dotenv

from langchain_openai import ChatOpenAI  # OpenAI-compatible client
from langchain.agents import initialize_agent, AgentType

from tarvos_mcp_tool import MCPTool  # Now using MCPTool which handles both Tarvos and demo tools


def main():
    # 1) Load environment variables from .env
    load_dotenv()

    # 2) Read environment variables (Tarvos gateway)
    api_url = os.getenv("TARVOS_BASE_URL")                
    api_key = os.getenv("TARVOS_API_KEY")                
    model_name = (
        os.getenv("SUMMARIZER_MODEL")
        or os.getenv("MODEL_NAME")
        or "meta-llama/Meta-Llama-3.1-8B-Instruct"
    )

    # 3) Validation
    missing = []
    if not api_url:
        missing.append("TARVOS_BASE_URL")
    if not api_key:
        missing.append("TARVOS_API_KEY")
    if missing:
        raise SystemExit(f"Missing environment variables: {', '.join(missing)}. Please configure them in .env.")

    # 4) Build the MCPTool (handles both Tarvos and demo tools)
    mcp_tool = MCPTool(api_url=api_url, api_key=api_key, model=model_name)

    # 5) Initialize LLM (also via Tarvos OpenAI-compatible chat endpoint)
    llm = ChatOpenAI(
        model=model_name,
        temperature=0,
        base_url=api_url,  
        api_key=api_key,
    )

    # 6) Register the tools and initialize the agent
    tarvos_tool = mcp_tool.as_tool("tarvos")  # Getting Tarvos tool
    demo_notes_tool = mcp_tool.as_tool("demo")  # Getting demo tool
    tools = [tarvos_tool, demo_notes_tool]  # Include both tools here
    agent = initialize_agent(
        tools=tools,
        llm=llm,
        agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
        verbose=True,
        max_iterations=1,  # limit loop iterations
    )

    # 7) Example call for summarizer tool (Tarvos Summarizer)
    print(">>> Starting MCP Demo: Calling Tarvos Summarizer Tool <<<")
    user_input = (
        "Summarize the following text in 3 sentences: "
        "Artificial Intelligence is the simulation of human intelligence "
        "processes by machines, especially computer systems."
    )
    result = agent.run(user_input)
    print("\n=== Agent Result ===")
    print(result)

    # 8) Handling user request for latest demo research notes
    print("\n>>> User request for latest demo research notes <<<")
    demo_note_input = "Please find the latest 3 research notes from demo."

    # AgentExecutor will now decide which tool to use based on the request
    demo_note_result = agent.run(demo_note_input)

    print("\n=== demo Research Notes Result ===")
    print(demo_note_result)


if __name__ == "__main__":
    main()

代码二：

from typing import Any
import logging
from fastmcp import FastMCP

USER_AGENT = "demo-app/1.0 (myemail@example.com)"
mcp = FastMCP(name="demo MCP server")


# Mock Data (replace these with your desired mock responses)
mock_contact_data = {
    "entity_id": "12345",
    "full_name": "Bill Lee",
    "email": "bill.lee@example.com",
    "phone": "555-1234",
}

mock_calendar_data = {
    "events": [
        {"event_id": "event1", "title": "Research Meeting", "date": "2025-09-30", "time": "10:00 AM"},
        {"event_id": "event2", "title": "Team Sync", "date": "2025-09-30", "time": "2:00 PM"}
    ]
}

mock_research_data = {
    "research_notes": [
        {"note_id": "note1", "title": "Research Progress", "date": "2025-09-25", "content": "Research is going well..."},
        {"note_id": "note2", "title": "Project Update", "date": "2025-09-24", "content": "Completed initial phase of project."}
    ]
}

# Mock Forward Request Function
async def forward_request(
        url: str,
        method: str = "GET",
        params: dict[str, Any] | None = None,
        data: dict[str, Any] | None = None,
        json: dict[str, Any] | None = None,
        formdata: dict[str, Any] | None = None,
        headers: dict[str, str] | None = None
) -> dict[str, Any] | None:
    """Mock forward a request to return mock data."""
    logging.info(f"Mock request to {url} with method {method} and params {params}")
    # You can return different mock data based on the URL or method
    if "contact" in url:
        return mock_contact_data
    elif "calendar" in url:
        return mock_calendar_data
    elif "research" in url:
        return mock_research_data
    else:
        return {"error": "Unknown endpoint"}

@mcp.tool(
    name="get_demo_contact",
    description="Get demo contact, you need to provide a username, then it will return mock contact data"
)
async def getdemoContact(username: str):
    # Log the incoming username for debugging
    logging.info(f"Received username for lookup: {username}")

    # Check if the username matches exactly "Bill Lee" (case-insensitive)
    if username.strip().lower() == "bill lee".lower():
        logging.info(f"Exact match found for username: {username}")
        return mock_contact_data
    else:
        logging.info(f"No exact match found for username: {username}")
        # Return an empty result or a suggestion to try different names
        return {
            "error": "No contact found. Try variations like 'bill_lee', 'billee', etc."
        }

@mcp.tool(
    name="get_demo_calendar",
    description="Get demo Calendar. If no contact ID is provided, it will use the default admin user ID."
)
async def getdemoCalendar(contactid: str = '23f03fe70a16aed0d7e210357164e401') -> dict[str, Any] | None:
    # Mocked response for calendar lookup
    logging.info(f"Fetching calendar for contact ID: {contactid}")
    return mock_calendar_data


@mcp.tool(
    name="get_demo_research",
    description="Get demo research or notes. If no contact ID is provided, it will use the default admin user ID."
)
async def getdemoNote(contactId: str = '23f03fe70a16aed0d7e210357164e401') -> dict[str, Any] | None:
    # Mocked response for research/notes lookup
    logging.info(f"Fetching research for contact ID: {contactId}")
    return mock_research_data


if __name__ == "__main__":
    mcp.run(transport="http", host="0.0.0.0", port=5000)