Agentic IDE 智能体协议解析与自定义

# 1. 安装 Node.js（如果尚未安装）
brew install node@22

# 2. 安装 Claude Code CLI
npm install -g @anthropic-ai/claude-code

# 3. 配置 API 密钥
export ANTHROPIC_API_KEY="sk-ant-..."

# 4. 验证安装
claude --version

# 创建项目目录
mkdir company-mcp && cd company-mcp

# 使用 uv 初始化项目
uv init

# 安装 fastmcp
uv add fastmcp

from fastmcp import FastMCP
import requests

mcp = FastMCP("Company Info MCP")

@mcp.tool()
def get_company_info(symbol: str) -> dict:
    """获取美股公司基本信息
    
    Args:
        symbol: 股票代码，如 AAPL、GOOGL
    
    Returns:
        公司名称、行业、市值等信息
    """
    url = f"https://api.example.com/stocks/{symbol}"
    response = requests.get(url)
    data = response.json()
    
    return {
        "name": data["name"],
        "sector": data["sector"],
        "marketCap": data["marketCap"],
        "employees": data["employees"]
    }

@mcp.tool()
def search_companies(keyword: str) -> list:
    """按关键词搜索公司
    
    Args:
        keyword: 搜索关键词
    
    Returns:
        匹配的公司列表
    """
    url = f"https://api.example.com/search?q={keyword}"
    response = requests.get(url)
    return response.json()["results"]

if __name__ == "__main__":
    mcp.run()

{
  "mcpServers": {
    "company-info": {
      "command": "uv",
      "args": ["run", "--directory", "/path/to/company-mcp", "python", "mcp_server.py"],
      "cwd": "/path/to/company-mcp"
    }
  }
}

$ claude

> 帮我查一下苹果公司的基本信息

[Claude Code 调用 get_company_info("AAPL")]

苹果公司的基本信息如下：
- 公司名称：Apple Inc.
- 所属行业：Technology
- 市值：2.8 万亿美元
- 员工数：164,000

> 再搜索一下所有 AI 相关的公司

[Claude Code 调用 search_companies("AI")]

找到以下公司：
1. OpenAI
2. Anthropic
3. Cohere
4. Hugging Face

{
  "name": "Search Agent",
  "description": "负责网络搜索和信息提取",
  "version": "1.0.0",
  "capabilities": {
    "search": {
      "description": "搜索网络获取信息",
      "inputSchema": {
        "query": { "type": "string" }
      },
      "outputSchema": {
        "results": { "type": "array" }
      }
    },
    "readUrl": {
      "description": "读取指定 URL 的内容",
      "inputSchema": {
        "url": { "type": "string" }
      },
      "outputSchema": {
        "content": { "type": "string" }
      }
    }
  },
  "endpoints": [
    {
      "url": "http://localhost:8001",
      "transport": "http"
    }
  ]
}

from fastapi import FastAPI
from pydantic import BaseModel
import httpx

app = FastAPI()

class SearchRequest(BaseModel):
    query: str

class ReadUrlRequest(BaseModel):
    url: str

@app.post("/search")
async def search(req: SearchRequest):
    """调用搜索 API"""
    async with httpx.AsyncClient() as client:
        resp = await client.get(
            "https://api.exa.ai/search",
            params={"query": req.query, "numResults": 5},
            headers={"Authorization": "Bearer YOUR_API_KEY"}
        )
    return {"results": resp.json()["results"]}

@app.post("/readUrl")
async def read_url(req: ReadUrlRequest):
    """读取网页内容"""
    async with httpx.AsyncClient() as client:
        resp = await client.get(req.url)
    return {"content": resp.text}

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

import httpx
import asyncio

class AgentOrchestrator:
    """智能体编排器"""
    
    def __init__(self):
        self.agents = {
            "search": "http://localhost:8001",
            "analysis": "http://localhost:8002",
            "writer": "http://localhost:8003"
        }
    
    async def call_agent(self, agent: str, action: str, data: dict):
        """调用指定智能体"""
        url = f"{self.agents[agent]}/{action}"
        async with httpx.AsyncClient() as client:
            resp = await client.post(url, json=data)
        return resp.json()
    
    async def generate_report(self, topic: str):
        """生成报告的完整工作流"""
        # Step 1: 搜索信息
        search_result = await self.call_agent(
            "search", "search", {"query": topic}
        )
        
        # Step 2: 读取关键网页
        content_results = []
        for item in search_result["results"][:3]:
            content = await self.call_agent(
                "search", "readUrl", {"url": item["url"]}
            )
            content_results.append(content["content"])
        
        # Step 3: 分析数据
        analysis = await self.call_agent(
            "analysis", "analyze", {"content": content_results}
        )
        
        # Step 4: 生成报告
        report = await self.call_agent(
            "writer", "write", {"analysis": analysis, "topic": topic}
        )
        
        return report

# 使用示例
async def main():
    orchestrator = AgentOrchestrator()
    report = await orchestrator.generate_report("AI Agent 发展趋势")
    print(report["content"])

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

#!/bin/bash

# 启动 Search Agent
uvicorn search_agent:app --port 8001 &
SEARCH_PID=$!

# 启动 Analysis Agent
uvicorn analysis_agent:app --port 8002 &
ANALYSIS_PID=$!

# 启动 Writer Agent
uvicorn writer_agent:app --port 8003 &
WRITER_PID=$!

# 启动编排器
python orchestrator.py

# 清理进程
kill $SEARCH_PID $ANALYSIS_PID $WRITER_PID