Agent 时代 MCP 已经过时了吗？

from mcp.server import FastMCP

mcp = FastMCP("Calculator Server")


@mcp.tool(
    description="""Add two numbers together and return their sum. 
    This is a fundamental arithmetic operation that takes two integer parameters (x and y) 
    and performs addition. The function will compute the mathematical sum of the two input values 
    and return the result as an integer. This tool is useful for basic calculations, 
    financial computations, data aggregation, and any scenario where you need to combine 
    two numeric values. The operation is commutative, meaning add(x, y) equals add(y, x). 
    Examples: add(5, 3) returns 8, add(100, 200) returns 300, add(-5, 10) returns 5."""
)
def add(x: int, y: int) -> int:
    """Add two numbers and return the result."""
    return x + y


mcp.run(transport="streamable-http")

import os

from mcp.client.streamable_http import streamable_http_client
from strands import Agent
from strands.models.openai import OpenAIModel
from strands.tools.mcp.mcp_client import MCPClient
from dotenv import load_dotenv


load_dotenv()

model = OpenAIModel(
    client_args={
        "base_url": os.getenv("DEEPSEEK_BASE_URL"),
        "api_key": os.getenv("DEEPSEEK_API_KEY"),
    },
    model_id="deepseek-chat",
)


def create_streamable_http_transport():
    return streamable_http_client("http://localhost:8000/mcp/")


streamable_http_mcp_client = MCPClient(create_streamable_http_transport)

# 使用 MCP Server 的上下文管理器（自动建立和关闭连接）
with streamable_http_mcp_client:
    # 从 MCP Server 获取可用工具列表
    tools = streamable_http_mcp_client.list_tools_sync()

    # 创建 Agent，并将 MCP 提供的工具注入 Agent
    agent = Agent(model=model, tools=tools)

    # 调用 Agent 执行任务
    agent("If I have 1000 and spend 246, how much do I have left?")

# hooks.py
from strands.hooks.events import (
    BeforeInvocationEvent,
    BeforeToolCallEvent,
    AfterToolCallEvent,
)

from tool_spec_manager import ToolSpecManager


custom_tool_descriptions = {"add": "加法"}

spec_manager = ToolSpecManager(custom_desc=custom_tool_descriptions)


def before_invocation_hook(event: BeforeInvocationEvent) -> None:
    """在每次 Agent 调用前精简所有工具规范"""

    spec_manager.minimize_tools(event.agent.tool_registry)


def before_tool_call_hook(event: BeforeToolCallEvent) -> None:
    """在工具调用前恢复完整规范"""

    if event.selected_tool:
        spec_manager.restore_tool(
            event.selected_tool.tool_name, event.agent.tool_registry
        )


def after_tool_call_hook(event: AfterToolCallEvent) -> None:
    """在工具调用后重新精简规范"""

    if event.selected_tool:
        spec_manager.minimize_tool(
            event.selected_tool.tool_name, event.agent.tool_registry
        )

from typing import Dict, Any, Optional
import json


class ToolSpecManager:
    def __init__(self, custom_desc: Optional[Dict[str, str]] = None):
        """初始化管理器

        Args:
            custom_desc: 将使用自定义描述代替自动生成的描述
        """

        self.original_specs: Dict[str, Dict[str, Any]] = {}
        self.minimized_specs: Dict[str, Dict[str, Any]] = {}
        self.custom_desc: Dict[str, str] = custom_desc or {}

    def minimize_tools(self, registry) -> None:
        """精简所有工具规范(包括 Tools/MCP 工具)"""

        for name, tool in registry.registry.items():
            if name not in self.original_specs:
                self.original_specs[name] = tool.tool_spec.copy()

            if name in self.custom_desc:
                description = self.custom_desc[name]
            else:
                original_desc = self.original_specs[name].get("description", "")
                description = (
                    original_desc[:50] + "..."
                    if len(original_desc) > 50
                    else original_desc
                )

            self.minimized_specs[name] = {
                "name": name,
                "description": description,
                "inputSchema": {
                    "json": {"type": "object", "properties": {}, "required": []}
                },
            }

            # MCP 工具：直接修改底层 mcp_tool 对象
            if hasattr(tool, "mcp_tool"):
                tool.mcp_tool.description = description
                tool.mcp_tool.inputSchema = {
                    "type": "object",
                    "properties": {},
                    "required": [],
                }
            else:
                # 普通工具：设置 tool_spec
                tool.tool_spec = self.minimized_specs[name]

    def restore_tool(self, tool_name: str, registry) -> bool:
        """恢复单个工具的完整规范"""

        if tool_name in self.original_specs:
            if tool_name in registry.registry:
                tool = registry.registry[tool_name]
                original_spec = self.original_specs[tool_name]

                # MCP 工具：修改底层 mcp_tool 对象
                if hasattr(tool, "mcp_tool"):
                    tool.mcp_tool.description = original_spec["description"]
                    tool.mcp_tool.inputSchema = original_spec["inputSchema"]["json"]
                else:
                    # 普通工具：设置 tool_spec
                    tool.tool_spec = original_spec

                return True
            else:
                return False
        else:
            return False

    def minimize_tool(self, tool_name: str, registry) -> bool:
        """重新精简单个工具的规范"""
        if tool_name in self.minimized_specs:
            if tool_name in registry.registry:
                tool = registry.registry[tool_name]
                minimized_spec = self.minimized_specs[tool_name]

                # MCP 工具：修改底层 mcp_tool 对象
                if hasattr(tool, "mcp_tool"):
                    tool.mcp_tool.description = minimized_spec["description"]
                    tool.mcp_tool.inputSchema = minimized_spec["inputSchema"]["json"]
                else:
                    # 普通工具：设置 tool_spec
                    tool.tool_spec = minimized_spec

                return True
            else:
                return False
        else:
            return False

    def get_stats(self) -> Dict[str, Any]:
        """获取优化统计信息"""

        if not self.original_specs:
            return {
                "total_tools": 0,
                "original_tokens_estimate": 0,
                "minimized_tokens_estimate": 0,
                "savings_estimate": 0,
                "savings_percentage": 0,
            }

        original_tokens = sum(
            len(json.dumps(spec)) // 4 for spec in self.original_specs.values()
        )
        minimized_tokens = sum(
            len(json.dumps(spec)) // 4 for spec in self.minimized_specs.values()
        )

        savings = original_tokens - minimized_tokens
        savings_pct = (savings / original_tokens * 100) if original_tokens > 0 else 0

        return {
            "total_tools": len(self.original_specs),
            "original_tokens_estimate": original_tokens,
            "minimized_tokens_estimate": minimized_tokens,
            "savings_estimate": savings,
            "savings_percentage": round(savings_pct, 2),
        }

import os
import json
from mcp.client.streamable_http import streamable_http_client
from strands import Agent
from strands.models.openai import OpenAIModel
from strands.tools.mcp.mcp_client import MCPClient
from strands.hooks.events import (
    BeforeInvocationEvent,
    BeforeToolCallEvent,
    AfterToolCallEvent,
)
from dotenv import load_dotenv

from hooks import (
    before_invocation_hook,
    before_tool_call_hook,
    after_tool_call_hook,
    spec_manager,
)

load_dotenv()


def estimate_tokens(text: str) -> int:
    """估算token数量"""
    return len(text) // 4


def get_tool_specs_size(tools):
    """计算工具规范大小"""

    specs_json = json.dumps([t.tool_spec for t in tools], ensure_ascii=False)
    return estimate_tokens(specs_json)


def print_tool_descriptions(tools, title="工具描述"):
    """打印工具描述信息"""
    print(f"\n{title}:")
    for tool in tools:
        spec = tool.tool_spec
        name = spec.get("name", "unknown")
        description = spec.get("description", "")
        input_schema = spec.get("inputSchema", {})

        print(f"  - {name}:")
        print(f"    描述: {description}")
        print(
            f"    参数数量: {len(input_schema.get('json', {}).get('properties', {}))}"
        )


def create_transport():
    return streamable_http_client("http://localhost:8000/mcp/")


test_question = "If I have 1000 and spend 246, how much do I have left?"


model = OpenAIModel(
    client_args={
        "base_url": os.getenv("DEEPSEEK_BASE_URL"),
        "api_key": os.getenv("DEEPSEEK_API_KEY"),
    },
    model_id="deepseek-chat",
)

# 场景1: 未优化
print("\n【场景1】未启用优化")
with MCPClient(create_transport) as mcp_client:
    tools = mcp_client.list_tools_sync()
    print(f"加载工具: {len(tools)} 个")

    before_tokens = get_tool_specs_size(tools)
    print(f"工具规范: ~{before_tokens} tokens")

    print_tool_descriptions(tools, "原始工具描述")

    agent = Agent(model=model, tools=tools)
    agent(test_question)

# 场景2: 启用优化
print("\n【场景2】启用优化")
with MCPClient(create_transport) as mcp_client:
    tools = mcp_client.list_tools_sync()
    print(f"加载工具: {len(tools)} 个")

    agent = Agent(model=model, tools=tools)

    agent.hooks.add_callback(BeforeInvocationEvent, before_invocation_hook)
    agent.hooks.add_callback(BeforeToolCallEvent, before_tool_call_hook)
    agent.hooks.add_callback(AfterToolCallEvent, after_tool_call_hook)

    class FakeEvent:
        def __init__(self, agent):
            self.agent = agent

    before_invocation_hook(FakeEvent(agent))  # type: ignore

    after_tokens = get_tool_specs_size(list(agent.tool_registry.registry.values()))
    print(f"工具规范: ~{after_tokens} tokens")

    print_tool_descriptions(
        list(agent.tool_registry.registry.values()), "精简后工具描述"
    )

    agent(test_question)

print("\n============ 优化效果对比 ============")
stats = spec_manager.get_stats()
saved = before_tokens - after_tokens
saved_pct = (saved / before_tokens * 100) if before_tokens > 0 else 0
print(f"优化前: ~{before_tokens} tokens")
print(f"优化后: ~{after_tokens} tokens")
print(f"节省: ~{saved} tokens ({saved_pct:.1f}%)")

from mcp.server import FastMCP
import time

mcp = FastMCP("Meeting Cloud Storage Server")


@mcp.tool(description="Download meeting minutes from internal system by meeting ID")
def download_meeting_minutes(meeting_id: str) -> dict:
    """Download meeting minutes from internal system.

    Args:
        meeting_id: Unique identifier of the meeting (e.g., "MTG-2026-001")

    Returns:
        dict: Meeting record with title, date, participants, and content
    """

    time.sleep(0.5)

    return {
        "meeting_id": meeting_id,
        "title": "Q4 Strategy Review Meeting",
        "date": "2026-01-10",
        "participants": ["Alice", "Bob", "Carol"],
        "summary": """Meeting Summary:

The Q4 Strategy Review Meeting was held on January 10, 2026, with key stakeholders from Product, Engineering, and Marketing teams. The meeting focused on three main areas:

1. Q3 Performance Analysis:
   - Revenue exceeded targets by 15%
   - User growth rate increased to 12% month-over-month
   - Customer satisfaction score improved to 4.6/5.0

2. Q4 Product Roadmap:
   - Priority 1: Launch AI-powered financial analysis features
   - Priority 2: Implement real-time collaboration tools
   - Priority 3: Expand mobile app capabilities

3. Budget and Resource Allocation:
   - Approved additional $50K budget specifically for AI feature development
   - Authorized hiring of 2 senior engineers to accelerate development
   - Committed to launching public beta program by November 2026

Key Decisions Made:
- Product team will deliver technical feasibility study by January 22
- Marketing will prepare beta program materials by January 25
- Engineering will finalize architecture design by January 20

Next Steps:
- Weekly progress reviews starting next Monday
- Monthly stakeholder updates
- Quarterly review scheduled for end of March 2026""",
        "status": "success",
    }


@mcp.tool(
    description="Upload meeting summary to third-party cloud storage (Google Drive, Dropbox, OneDrive, etc.)"
)
def upload_to_cloud(
    cloud_service: str,
    summary_content: str,
    meeting_id: str,
) -> dict:
    """Upload meeting summary to third-party cloud storage.

    Args:
        cloud_service: Target cloud service (e.g., "google_drive", "dropbox", "onedrive")
        summary_content: Meeting summary content to upload
        meeting_id: Meeting identifier for file naming

    Returns:
        dict: Upload result with file URL and metadata
    """

    time.sleep(0.8)

    return {
        "status": "success",
        "cloud_service": cloud_service,
        "file_name": f"{meeting_id}_summary.txt",
        "content_size": len(summary_content),
        "file_url": f"https://{cloud_service}.com/files/{meeting_id}",
    }


mcp.run(transport="streamable-http")

import os

from mcp.client.streamable_http import streamable_http_client
from strands import Agent
from strands.models.openai import OpenAIModel
from strands.tools.mcp.mcp_client import MCPClient
from dotenv import load_dotenv

load_dotenv()

model = OpenAIModel(
    client_args={
        "base_url": os.getenv("DEEPSEEK_BASE_URL"),
        "api_key": os.getenv("DEEPSEEK_API_KEY"),
    },
    model_id="deepseek-chat",
)


def create_streamable_http_transport():
    return streamable_http_client("http://localhost:8000/mcp/")


streamable_http_mcp_client = MCPClient(create_streamable_http_transport)

with streamable_http_mcp_client:
    tools = streamable_http_mcp_client.list_tools_sync()

    print(f"加载了 {len(tools)} 个工具:")
    for tool in tools:
        print(f"  - {tool.tool_spec['name']}: {tool.tool_spec['description'][:60]}...")

    agent = Agent(model=model, tools=tools)

    result = agent(
        "请帮我下载会议ID为 MTG-2026-001 的会议记录，"
        "然后将会议摘要上传到 Google Drive。"
    )

    print("Token 使用情况:\n")
    if result.metrics:
        usage = result.metrics.accumulated_usage
        print(f"输入 tokens: {usage.get('inputTokens', 0)}")
        print(f"输出 tokens: {usage.get('outputTokens', 0)}")
        print(f"总计 tokens: {usage.get('totalTokens', 0)}")

import json

from mcp.client.streamable_http import streamable_http_client
from strands.tools.mcp.mcp_client import MCPClient
from strands import tool


def create_streamable_http_transport():
    return streamable_http_client("http://localhost:8000/mcp/")


mcp_client = MCPClient(create_streamable_http_transport)


def download_and_upload(meeting_id: str, cloud_service: str):
    """Internal workflow function that calls remote MCP tools."""

    meeting_result = mcp_client.call_tool_sync(
        tool_use_id="workflow-step-1",
        name="download_meeting_minutes",
        arguments={"meeting_id": meeting_id},
    )

    meeting_data = json.loads(meeting_result["content"][0]["text"])

    if meeting_data.get("status") != "success":
        return {
            "status": "error",
            "message": "Failed to download meeting minutes",
            "meeting_id": meeting_id,
        }

    upload_result_raw = mcp_client.call_tool_sync(
        tool_use_id="workflow-step-2",
        name="upload_to_cloud",
        arguments={
            "cloud_service": cloud_service,
            "summary_content": meeting_data["summary"],
            "meeting_id": meeting_id,
        },
    )

    upload_result = json.loads(upload_result_raw["content"][0]["text"])

    return {
        "status": "success",
        "file_url": upload_result["file_url"],
        "message": f"Downloaded '{meeting_data['title']}' and uploaded to {cloud_service}",
    }


@tool
def download_and_upload_meeting(
    meeting_id: str,
    cloud_service: str = "google_drive",
) -> dict:
    """Download meeting minutes and upload to cloud storage in one operation.

    This is a client-side workflow tool that orchestrates multiple MCP tool calls,
    reducing the number of AI calls and improving reliability.

    Args:
        meeting_id: Unique identifier of the meeting (e.g., "MTG-2026-001")
        cloud_service: Target cloud service (default: "google_drive", options: "dropbox", "onedrive")

    Returns:
        dict: Result with status, file_url, and message
    """
    return download_and_upload(meeting_id, cloud_service)

import os
from strands import Agent
from strands.models.openai import OpenAIModel
from dotenv import load_dotenv

from tools import download_and_upload_meeting, mcp_client

load_dotenv()

model = OpenAIModel(
    client_args={
        "base_url": os.getenv("DEEPSEEK_BASE_URL"),
        "api_key": os.getenv("DEEPSEEK_API_KEY"),
    },
    model_id="deepseek-chat",
)


with mcp_client:
    agent = Agent(model=model, tools=[download_and_upload_meeting])

    result = agent(
        "请帮我下载会议ID为 MTG-2026-001 的会议记录，"
        "然后将会议摘要上传到 Google Drive。"
    )

    print("\nToken 使用情况:")
    if result.metrics:
        usage = result.metrics.accumulated_usage
        print(f"  输入 tokens: {usage.get('inputTokens', 0)}")
        print(f"  输出 tokens: {usage.get('outputTokens', 0)}")
        print(f"  总计 tokens: {usage.get('totalTokens', 0)}")