1
Client → Server：查一下今天成都天气
2
Server → Client：返回结果

1
Client → Server：生成一份报告
2
Server → Client：正在检索资料
3
Server → Client：正在生成大纲
4
Server → Client：正在生成正文
5
Server → Client：完成

1
Client：你慢慢跑，跑完通知我
2
Server：收到
3
……
4
Server → webhook：任务完成

1
Burger Seller Agent 启动
2
  ↓
3
暴露 Agent Card：/.well-known/agent.json
4
  ↓
5
购买助理 Agent 启动
6
  ↓
7
读取远程 Agent Card，保存到 cards / remote_agent_connections
8
  ↓
9
用户说：我想买汉堡
10
  ↓
11
购买助理根据 Agent Card 判断该找 Burger Seller Agent
12
  ↓
13
购买助理把用户请求封装成 Message
14
  ↓
15
通过 A2A Client 发给 Burger Seller Agent
16
  ↓
17
Burger Seller Agent 收到请求，形成 Task 或直接返回 Message
18
  ↓
19
Burger Seller Agent 的 AgentExecutor 执行业务逻辑
20
  ↓
21
返回确认问题 / 订单结果 / Artifact
22
  ↓
23
购买助理把结果转给用户

1
sell_burger_skill = AgentSkill(
2
    id="sell_burger",
3
    name="Sell Burger",
4
    description="Handle burger ordering requests",
5
    tags=["burger", "food", "order"],
6
    examples=[
7
        "I want a cheeseburger",
8
        "我想买一个汉堡"
9
    ]
10
)

1
agent_card = AgentCard(
2
    name="Burger Seller Agent",
3
    description="A seller agent that handles burger purchase requests",
4
    url="http://localhost:10001/",
5
    version="1.0.0",
6
    skills=[sell_burger_skill],
7
)

1
{
2
  "name": "Burger Seller Agent",
3
  "description": "A seller agent that handles burger purchase requests",
4
  "url": "http://localhost:10001/",
5
  "skills": [
6
    {
7
      "id": "sell_burger",
8
      "name": "Sell Burger",
9
      "description": "Handle burger ordering requests"
10
    }
11
  ]
12
}

1
remote_agent_addresses = [
2
    "http://localhost:10001",  # Burger Seller Agent
3
    "http://localhost:10002",  # Pizza Seller Agent
4
]

1
for address in remote_agent_addresses:
2
    card = get(f"{address}/.well-known/agent.json")
3
    cards[address] = card

1
{
2
  "role": "user",
3
  "parts": [
4
    {
5
      "kind": "text",
6
      "text": "Order lunch for the team."
7
    },
8
    {
9
      "kind": "data",
10
      "data": {
11
        "people": 5,
12
        "budget": 80,
13
        "dietaryRestrictions": ["no pork", "one vegetarian"]
14
      }
15
    }
16
  ],
17
  "messageId": "msg-001"
18
}

1
class BurgerAgentExecutor(AgentExecutor):
2
    async def execute(self, context, event_queue):
3
        # 1. 从 context 中取出购买助理发来的 Message
4
        user_message = context.message
5

6
        # 2. 调用自己的业务逻辑，例如检查菜单、计算价格、处理饮食限制
7
        result = await burger_seller.handle_order(user_message)
8

9
        # 3. 把处理结果通过 event_queue 返回给调用方
10
        await event_queue.enqueue_event(
11
            new_agent_text_message(result)
12
        )

1
I can order 4 cheeseburgers and 1 veggie burger for 78 dollars. Please confirm.

1
submitted → working → input_required → working → completed

1
{
2
  "id": "task-123",
3
  "status": {
4
    "state": "input_required",
5
    "message": {
6
      "role": "agent",
7
      "parts": [
8
        {
9
          "kind": "text",
10
          "text": "I can order 4 cheeseburgers and 1 veggie burger for 78 dollars. Please confirm."
11
        }
12
      ]
13
    }
14
  }
15
}

1
{
2
  "artifactId": "receipt-001",
3
  "name": "burger-order-receipt",
4
  "parts": [
5
    {
6
      "kind": "data",
7
      "data": {
8
        "orderId": "B123",
9
        "items": ["4 cheeseburgers", "1 veggie burger"],
10
        "total": 78,
11
        "status": "confirmed"
12
      }
13
    }
14
  ]
15
}

1
服务端 Agent 要做：
2
定义 AgentSkill / AgentCard，配置 TaskStore，实现 AgentExecutor，并借助 A2A SDK 提供的 A2A Server 将自己暴露成可远程调用的 Agent。
3

4
客户端 Agent 要做：
5
获取远程 Agent 地址，读取 Agent Card，根据 Agent Card 选择合适的远程 Agent，构造 Message，发送 A2A 请求，并处理返回的 Message / Task 状态 / Artifact。

1
import os
2

3
import uvicorn
4

5
from a2a.helpers import (
6
    get_message_text,
7
    new_task_from_user_message,
8
    new_text_message,
9
    new_text_part,
10
)
11
from a2a.server.agent_execution import AgentExecutor, RequestContext
12
from a2a.server.events import EventQueue
13
from a2a.server.request_handlers import DefaultRequestHandler
14
from a2a.server.routes import create_agent_card_routes, create_jsonrpc_routes
15
from a2a.server.tasks import InMemoryTaskStore, TaskUpdater
16
from a2a.types import AgentCapabilities, AgentCard, AgentInterface, AgentSkill
17
from a2a.types.a2a_pb2 import TaskState
18
from starlette.applications import Starlette
19

20

21
A2A_HOST = os.getenv("A2A_HOST", "127.0.0.1")
22
A2A_PORT = int(os.getenv("A2A_PORT", "10001"))
23

24

25
class BurgerSeller:
26
    """真正的业务逻辑。
27

28
    A2A 只负责 Agent 之间如何发现、发消息、维护任务状态。
29
    至于“卖汉堡”这件事本身怎么实现，属于业务代码。
30
    """
31

32
    async def prepare_quote(self, user_request: str) -> str:
33
        """根据用户请求生成报价。
34

35
        这里简化为直接返回固定报价。
36
        真实项目里，这一步可能会查库存、算价格、调用支付/餐厅系统等。
37
        """
38
        return (
39
            "I can order 4 cheeseburgers and 1 veggie burger "
40
            "for 78 dollars. Please confirm."
41
        )
42

43
    async def place_order(self) -> str:
44
        """用户确认后真正下单。"""
45
        return (
46
            "Order confirmed. I placed the burger order and the restaurant "
47
            "will deliver it in about 30 minutes."
48
        )
49

50

51
class BurgerAgentExecutor(AgentExecutor):
52
    """A2A 服务端真正处理消息的入口。
53
    每次客户端发送 Message，A2A SDK 都会调用 execute()。
54
    """
55

56
    def __init__(self) -> None:
57
        self.seller = BurgerSeller()
58

59
    async def execute(
60
        self,
61
        context: RequestContext,
62
        event_queue: EventQueue,
63
    ) -> None:
64
        """处理一次 A2A message:send 请求。
65

66
        context 里包含本次请求的 Message、当前 Task、context_id 等信息。
67
        event_queue 是服务端向 A2A 框架发布 Task 变化的通道。
68
        """
69
        # 如果客户端传来了已有 task_id，A2A SDK 会从 TaskStore 中恢复 current_task。
70
        # 这就是“长任务可以继续”的关键：第二次确认不是新任务，而是同一个任务。
71
        if context.current_task:
72
            task = context.current_task
73
        else:
74
            # 第一次请求还没有 Task，需要从用户 Message 创建一个新的 Task。
75
            # new_task_from_user_message 会生成 task_id/context_id，并把用户消息放入 history。
76
            task = new_task_from_user_message(context.message)
77
            await event_queue.enqueue_event(task)
78

79
        # TaskUpdater 是 SDK 提供的辅助类，用来发布状态更新和 artifact。
80
        # 后面所有 WORKING、INPUT_REQUIRED、COMPLETED 都通过它写回同一个 Task。
81
        task_updater = TaskUpdater(
82
            event_queue=event_queue,
83
            task_id=task.id,
84
            context_id=task.context_id,
85
        )
86

87
        # A2A Message 可以有多个 part，这里只演示 text/plain，所以直接提取文本。
88
        user_text = get_message_text(context.message).strip()
89

90
        # 如果当前 Task 已经处于 INPUT_REQUIRED，说明上一次服务端在等待用户输入。
91
        # 此时这条新消息就应该被当作“用户确认/补充信息”，而不是新订单。
92
        if task.status.state == TaskState.TASK_STATE_INPUT_REQUIRED:
93
            await self._handle_confirmation(user_text, task_updater)
94
            return
95

96
        # 第一次处理订单：先把任务状态改成 WORKING，表示 Agent 正在处理。
97
        await task_updater.update_status(
98
            state=TaskState.TASK_STATE_WORKING,
99
            message=new_text_message("Burger Seller Agent is preparing a quote..."),
100
        )
101

102
        # 生成报价，并把报价作为 artifact 写入 Task。
103
        # artifact 更像任务产物；status.message 更像当前状态说明。
104
        quote = await self.seller.prepare_quote(user_text)
105
        await task_updater.add_artifact(
106
            parts=[
107
                new_text_part(
108
                    text=quote,
109
                    media_type="text/plain",
110
                )
111
            ]
112
        )
113

114
        # INPUT_REQUIRED 表示任务被中断/暂停，正在等待用户继续提供输入。
115
        # 客户端拿到这个状态后，可以提示用户输入 confirm，再带着同一个 task_id 续发消息。
116
        await task_updater.requires_input(
117
            message=new_text_message("Please confirm if you want me to place this order."),
118
        )
119

120
    async def _handle_confirmation(
121
        self,
122
        user_text: str,
123
        task_updater: TaskUpdater,
124
    ) -> None:
125
        """处理 INPUT_REQUIRED 状态下用户发来的确认消息。"""
126
        if user_text.lower() not in {"confirm", "yes", "y", "ok", "sure"}:
127
            await task_updater.requires_input(
128
                message=new_text_message(
129
                    "Please reply with 'confirm' to place the order."
130
                ),
131
            )
132
            return
133

134
        # 用户确认后，任务从 INPUT_REQUIRED 回到 WORKING。
135
        await task_updater.update_status(
136
            state=TaskState.TASK_STATE_WORKING,
137
            message=new_text_message("Confirmation received. Placing the order..."),
138
        )
139

140
        # 执行真正的下单动作，并把最终结果追加成新的 artifact。
141
        # 所以客户端最终会看到两个 artifact：报价结果 + 下单结果。
142
        result = await self.seller.place_order()
143
        await task_updater.add_artifact(
144
            parts=[
145
                new_text_part(
146
                    text=result,
147
                    media_type="text/plain",
148
                )
149
            ]
150
        )
151

152
        # 任务完成后进入终态 COMPLETED。
153
        # 进入终态后，这个 Task 就不应该再继续处理新的业务步骤。
154
        await task_updater.update_status(
155
            state=TaskState.TASK_STATE_COMPLETED,
156
            message=new_text_message("Order task is completed."),
157
        )
158

159
    async def cancel(
160
        self,
161
        context: RequestContext,
162
        event_queue: EventQueue,
163
    ) -> None:
164
        # 这里暂时没有实现取消逻辑。
165
        raise NotImplementedError("Cancel is not supported.")
166

167

168
def build_app() -> Starlette:
169
    """创建 A2A HTTP 服务。
170

171
    这里使用 Starlette 承载 A2A SDK 生成的路由。
172
    对外暴露两类能力：
173
    1. Agent Card 路由：让客户端发现这个 Agent 是谁、会什么、怎么访问。
174
    2. JSON-RPC 路由：让客户端发送 A2A message:send 请求。
175
    """
176
    # AgentSkill 描述这个 Agent 的一个具体能力。
177
    # 客户端或调度系统可以通过 Skill 判断是否应该把某类任务发给这个 Agent。
178
    sell_burger_skill = AgentSkill(
179
        id="sell_burger",
180
        name="Sell Burger",
181
        description="Handle burger ordering requests with manual confirmation",
182
        input_modes=["text/plain"],
183
        output_modes=["text/plain"],
184
        tags=["burger", "food", "order"],
185
        examples=[
186
            "I want a cheeseburger",
187
            "Order lunch for the team",
188
            "Confirm the burger order",
189
        ],
190
    )
191

192
    # AgentCard 是 A2A 的“名片”。
193
    # 客户端通常先读取 AgentCard，再根据里面的协议地址、输入输出类型、能力描述来创建客户端。
194
    public_agent_card = AgentCard(
195
        name="Burger Seller Agent",
196
        description="A seller agent that quotes and confirms burger purchase requests.",
197
        version="1.0.0",
198
        default_input_modes=["text/plain"],
199
        default_output_modes=["text/plain"],
200
        capabilities=AgentCapabilities(streaming=False),
201
        supported_interfaces=[
202
            AgentInterface(
203
                protocol_binding="JSONRPC",
204
                url=f"http://{A2A_HOST}:{A2A_PORT}",
205
            )
206
        ],
207
        skills=[sell_burger_skill],
208
    )
209

210
    # RequestHandler 把协议层和业务执行器串起来。
211
    # InMemoryTaskStore 用内存保存 Task，所以服务端重启后历史任务会丢失。
212
    # 真实系统里通常会换成数据库或持久化 TaskStore。
213
    request_handler = DefaultRequestHandler(
214
        agent_executor=BurgerAgentExecutor(),
215
        task_store=InMemoryTaskStore(),
216
        agent_card=public_agent_card,
217
    )
218

219
    # create_agent_card_routes 暴露 /.well-known/agent-card.json 等发现路由。
220
    # create_jsonrpc_routes 暴露 JSON-RPC 请求入口，本 demo 绑定在根路径 "/"。
221
    routes = []
222
    routes.extend(create_agent_card_routes(public_agent_card))
223
    routes.extend(create_jsonrpc_routes(request_handler, "/"))
224

225
    return Starlette(routes=routes)
226

227

228
if __name__ == "__main__":
229
    app = build_app()
230
    uvicorn.run(app, host=A2A_HOST, port=A2A_PORT)

1
import asyncio
2
import os
3

4
import httpx
5
from a2a.client import A2ACardResolver, ClientConfig, create_client
6
from a2a.helpers import new_text_message
7
from a2a.types.a2a_pb2 import Role, SendMessageRequest, StreamResponse, Task, TaskState
8

9

10
A2A_HOST = os.getenv("A2A_HOST", "127.0.0.1")
11
A2A_PORT = int(os.getenv("A2A_PORT", "10001"))
12
BURGER_AGENT_URL = f"http://{A2A_HOST}:{A2A_PORT}"
13

14

15
async def get_agent_card():
16
    """读取远程 Agent Card"""
17
    async with httpx.AsyncClient() as httpx_client:
18
        resolver = A2ACardResolver(
19
            httpx_client=httpx_client,
20
            base_url=BURGER_AGENT_URL,
21
        )
22
        return await resolver.get_agent_card()
23

24

25
def get_artifact_text(task: Task) -> list[str]:
26
    """从 Task 的 artifacts 中提取文本结果。"""
27
    texts = []
28
    for artifact in task.artifacts:
29
        for part in artifact.parts:
30
            if part.text:
31
                texts.append(part.text)
32
    return texts
33

34

35
def print_task_summary(task: Task) -> None:
36
    """
37
    SDK 返回的原始 protobuf 对象信息很完整，但有些繁杂。
38
    这里提取 task_id、context_id、状态、状态消息和 artifacts，方便观察长任务流转。
39
    """
40
    state_name = TaskState.Name(task.status.state)
41
    print(f"\nTask: {task.id}")
42
    print(f"Context: {task.context_id}")
43
    print(f"State: {state_name}")
44

45
    # status.message 表示“当前 Task 状态附带的一句话说明”。
46
    if task.status.message.parts:
47
        print("Status message:")
48
        for part in task.status.message.parts:
49
            if part.text:
50
                print(f"  {part.text}")
51

52
    # artifacts 表示“任务已经产生的结果”。
53
    # 第一次请求会产生报价 artifact；确认后会追加下单成功 artifact。
54
    artifact_texts = get_artifact_text(task)
55
    if artifact_texts:
56
        print("Artifacts:")
57
        for text in artifact_texts:
58
            print(f"  {text}")
59

60

61
async def send_message(
62
    client,
63
    user_text: str,
64
    task_id: str | None = None,
65
    context_id: str | None = None,
66
) -> Task:
67
    """向 A2A Agent 发送一条用户消息，并返回服务端给出的 Task。
68

69
    第一次发送消息时不传 task_id/context_id，服务端会创建新 Task。
70
    当服务端返回 INPUT_REQUIRED 后，第二次发送确认消息时必须传回同一个
71
    task_id/context_id，这样服务端才知道是在继续同一个长任务。
72
    """
73
    message = new_text_message(
74
        text=user_text,
75
        role=Role.ROLE_USER,
76
        task_id=task_id,
77
        context_id=context_id,
78
    )
79
    request = SendMessageRequest(message=message)
80

81
    # client.send_message 返回的是一个异步迭代器。
82
    # 即使当前配置是非流式，SDK 也会把结果包装成 StreamResponse，保持统一接口。
83
    last_task = None
84
    async for chunk in client.send_message(request):
85
        if isinstance(chunk, Task):
86
            last_task = chunk
87
        elif isinstance(chunk, StreamResponse) and chunk.HasField("task"):
88
            last_task = chunk.task
89

90
    if last_task is None:
91
        raise RuntimeError("The agent did not return a task.")
92

93
    return last_task
94

95

96
async def send_order_message(user_text: str):
97
    """完整演示一次“报价 -> 等待确认 -> 继续完成”的 A2A 长任务。"""
98
    agent_card = await get_agent_card()
99

100
    print("Read remote Agent Card:")
101
    print(f"- name: {agent_card.name}")
102
    print(f"- description: {agent_card.description}")
103

104
    # 服务端每次请求返回一个聚合后的 Task，而不是持续推送事件流。
105
    config = ClientConfig(streaming=False)
106
    client = await create_client(
107
        agent=agent_card,
108
        client_config=config,
109
    )
110

111
    try:
112
        # 第一次发送订单请求。
113
        # 服务端会创建 Task，生成报价 artifact，然后把状态置为 INPUT_REQUIRED。
114
        print("\nSending A2A message:")
115
        print(user_text)
116
        task = await send_message(client, user_text)
117
        print_task_summary(task)
118

119
        # 只要服务端说 INPUT_REQUIRED，客户端就应该把控制权交还给用户。
120
        # 用户输入 confirm 后，客户端带着同一个 task_id/context_id 继续发消息。
121
        while task.status.state == TaskState.TASK_STATE_INPUT_REQUIRED:
122
            confirmation = input("\nType 'confirm' to continue this A2A task: ").strip()
123
            task = await send_message(
124
                client,
125
                confirmation,
126
                task_id=task.id,
127
                context_id=task.context_id,
128
            )
129
            print_task_summary(task)
130
    finally:
131
        # 显式关闭客户端底层 httpx 连接。
132
        await client.close()
133

134

135
async def main():
136
    # 这个文本就是用户的自然语言任务请求。
137
    user_text = (
138
        "Order lunch for the team. "
139
        "people=5, budget=80, dietaryRestrictions=no pork and one vegetarian."
140
    )
141

142
    await send_order_message(user_text)
143

144

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

1
import json
2
import os
3
import smtplib
4
from datetime import datetime
5
from email.message import EmailMessage
6
from pathlib import Path
7

8
import httpx
9
from dotenv import load_dotenv
10
from mcp.server.fastmcp import FastMCP
11
from openai import OpenAI
12

13

14
BASE_DIR = Path(__file__).resolve().parent
15
GOOGLE_NEWS_DIR = BASE_DIR / "google_news"
16
SENTIMENT_REPORTS_DIR = BASE_DIR / "sentiment_reports"
17

18
# 加载环境变量。
19
load_dotenv()
20

21
# 初始化 MCP 服务器
22
mcp = FastMCP("NewsServer")
23

24

25
def _safe_filename(filename: str, default_prefix: str, suffix: str) -> str:
26
    """把模型或用户给出的文件名清理成 Windows/Linux 都安全的文件名。"""
27
    filename = filename.strip()
28
    if not filename:
29
        filename = f"{default_prefix}_{datetime.now().strftime('%Y%m%d_%H%M%S')}{suffix}"
30

31
    filename = os.path.basename(filename)
32
    filename = "".join(ch for ch in filename if ch not in '\\/:*?"<>|')
33

34
    if not filename.endswith(suffix):
35
        filename = f"{filename}{suffix}"
36

37
    return filename
38

39

40
def _resolve_report_path(filename: str = "", attachment_path: str = "") -> Path:
41
    """把邮件工具收到的文件名或路径解析成真实报告路径。"""
42
    raw_path = attachment_path or filename
43
    if not raw_path:
44
        raise ValueError("filename 或 attachment_path 不能为空")
45

46
    path = Path(raw_path)
47
    if path.is_absolute():
48
        return path
49

50
    # 如果传入的是 sentiment_reports/xxx.md 这种相对路径，就按 MCP 目录解析。
51
    candidate = BASE_DIR / path
52
    if candidate.exists():
53
        return candidate
54

55
    # 如果只传入 xxx.md，就默认它在 sentiment_reports 目录下。
56
    return SENTIMENT_REPORTS_DIR / path.name
57

58

59
@mcp.tool()
60
async def search_google_news(keyword: str) -> str:
61
    """使用 Serper API 搜索 Google News，并返回前 5 条新闻。
62

63
    参数:
64
        keyword: 新闻搜索关键词，例如“小米汽车”。
65

66
    返回:
67
        JSON 字符串，包含标题、摘要和链接；同时会在本地保存一份 JSON 文件。
68
    """
69
    api_key = os.getenv("SERPER_API_KEY")
70
    if not api_key:
71
        return "未配置 SERPER_API_KEY，请在 .env 文件中设置。"
72

73
    url = "https://google.serper.dev/news"
74
    headers = {
75
        "X-API-KEY": api_key,
76
        "Content-Type": "application/json",
77
    }
78
    payload = {"q": keyword}
79

80
    async with httpx.AsyncClient() as client:
81
        response = await client.post(url, headers=headers, json=payload)
82
        response.raise_for_status()
83
        data = response.json()
84

85
    if "news" not in data:
86
        return "未获取到搜索结果。"
87

88
    articles = [
89
        {
90
            "title": item.get("title"),
91
            "desc": item.get("snippet"),
92
            "url": item.get("link"),
93
        }
94
        for item in data["news"][:5]
95
    ]
96

97
    GOOGLE_NEWS_DIR.mkdir(parents=True, exist_ok=True)
98
    filename = f"google_news_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json"
99
    file_path = GOOGLE_NEWS_DIR / filename
100

101
    file_path.write_text(
102
        json.dumps(articles, ensure_ascii=False, indent=2),
103
        encoding="utf-8",
104
    )
105

106
    return (
107
        f"已获取与 [{keyword}] 相关的前 5 条 Google 新闻：\n"
108
        f"{json.dumps(articles, ensure_ascii=False, indent=2)}\n"
109
        f"已保存到：{file_path}"
110
    )
111

112

113
@mcp.tool()
114
async def analyze_sentiment(text: str, filename: str = "") -> str:
115
    """对新闻文本做舆情分析，并保存成 Markdown 报告。
116

117
    参数:
118
        text: 新闻内容或搜索工具返回的新闻 JSON。
119
        filename: 希望保存的 Markdown 文件名。可以不传，不传时自动生成。
120

121
    返回:
122
        报告文件的完整路径。后续 send_email_with_attachment 可以直接使用这个路径。
123
    """
124
    openai_key = os.getenv("DASHSCOPE_API_KEY")
125
    model = os.getenv("MODEL")
126
    base_url = os.getenv("BASE_URL")
127

128
    if not openai_key:
129
        return "未配置 DASHSCOPE_API_KEY，请在 .env 文件中设置。"
130
    if not model:
131
        return "未配置 MODEL，请在 .env 文件中设置。"
132

133
    client = OpenAI(api_key=openai_key, base_url=base_url)
134

135
    prompt = (
136
        "请对以下新闻内容进行舆情分析，要求包含：\n"
137
        "1. 总体情绪倾向\n"
138
        "2. 正面因素\n"
139
        "3. 负面或风险因素\n"
140
        "4. 对企业或品牌的影响\n"
141
        "5. 简短结论\n\n"
142
        f"{text}"
143
    )
144

145
    response = client.chat.completions.create(
146
        model=model,
147
        messages=[{"role": "user", "content": prompt}],
148
    )
149
    result = response.choices[0].message.content.strip()
150

151
    markdown = f"""# 舆情分析报告
152

153
**分析时间：** {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}
154

155
---
156

157
## 原始新闻材料
158

159
{text}
160

161
---
162

163
## 分析结果
164

165
{result}
166
"""
167

168
    SENTIMENT_REPORTS_DIR.mkdir(parents=True, exist_ok=True)
169

170
    # 不管 filename 是否为空，都必须生成 file_path。
171
    safe_name = _safe_filename(filename, "sentiment", ".md")
172
    file_path = SENTIMENT_REPORTS_DIR / safe_name
173
    file_path.write_text(markdown, encoding="utf-8")
174

175
    return str(file_path)
176

177

178
@mcp.tool()
179
async def send_email_with_attachment(
180
    to: str,
181
    subject: str,
182
    body: str,
183
    filename: str = "",
184
    attachment_path: str = "",
185
) -> str:
186
    """发送带附件的邮件。
187

188
    参数:
189
        to: 收件人邮箱。
190
        subject: 邮件标题。
191
        body: 邮件正文。
192
        filename: sentiment_reports 目录下的报告文件名。
193
        attachment_path: 报告文件完整路径。若传了这个参数，会优先使用它。
194

195
    返回:
196
        邮件发送结果说明。
197
    """
198
    smtp_server = os.getenv("SMTP_SERVER")
199
    smtp_port = int(os.getenv("SMTP_PORT", "465"))
200
    sender_email = os.getenv("EMAIL_USER")
201
    sender_pass = os.getenv("EMAIL_PASS")
202

203
    if not smtp_server or not sender_email or not sender_pass:
204
        return "SMTP 配置不完整，请检查 SMTP_SERVER、EMAIL_USER、EMAIL_PASS。"
205

206
    try:
207
        full_path = _resolve_report_path(filename=filename, attachment_path=attachment_path)
208
    except ValueError as exc:
209
        return f"附件参数无效：{exc}"
210

211
    if not full_path.exists():
212
        return f"附件路径无效，未找到文件：{full_path}"
213

214
    msg = EmailMessage()
215
    msg["Subject"] = subject
216
    msg["From"] = sender_email
217
    msg["To"] = to
218
    msg.set_content(body)
219

220
    try:
221
        file_data = full_path.read_bytes()
222
        msg.add_attachment(
223
            file_data,
224
            maintype="application",
225
            subtype="octet-stream",
226
            filename=full_path.name,
227
        )
228
    except Exception as exc:
229
        return f"附件读取失败：{exc}"
230

231
    try:
232
        with smtplib.SMTP_SSL(smtp_server, smtp_port) as server:
233
            server.login(sender_email, sender_pass)
234
            server.send_message(msg)
235
        return f"邮件已成功发送给 {to}，附件路径：{full_path}"
236
    except Exception as exc:
237
        return f"邮件发送失败：{exc}"
238

239

240
if __name__ == "__main__":
241
    mcp.run(transport="stdio")

1
import asyncio
2
import os
3
import json
4
from typing import Optional, List
5
from contextlib import AsyncExitStack
6
from datetime import datetime
7
import re
8
from openai import OpenAI
9
from dotenv import load_dotenv
10
from mcp import ClientSession, StdioServerParameters
11
from mcp.client.stdio import stdio_client
12

13
load_dotenv()
14

15
class MCPClient:
16
  # 基础依赖层：统一管理全局资源、大模型客户端、资源自动回收栈
17
    def __init__(self):
18
        self.exit_stack = AsyncExitStack()  # 异步资源管理器（核心）
19
        self.openai_api_key = os.getenv("DASHSCOPE_API_KEY") # 读取.env大模型配置
20
        self.base_url = os.getenv("BASE_URL")
21
        self.model = os.getenv("MODEL")
22
        if not self.openai_api_key:
23
            raise ValueError("❌ 未找到 OpenAI API Key，请在 .env 文件中设置 DASHSCOPE_API_KEY")
24
        self.client = OpenAI(api_key=self.openai_api_key, base_url=self.base_url)  # 初始化大模型客户端
25
        self.session: Optional[ClientSession] = None # MCP会话句柄
26

27
  # MCP 服务连接层：启动 MCP 服务子进程、建立双向 stdio 通信、初始化 MCP 会话、拉取服务端可用工具列表
28
    async def connect_to_server(self, server_script_path: str):
29
        # 对服务器脚本进行判断，只允许是 .py 或 .js
30
        is_python = server_script_path.endswith('.py')
31
        is_js = server_script_path.endswith('.js')
32
        if not (is_python or is_js):
33
            raise ValueError("服务器脚本必须是 .py 或 .js 文件")
34

35
        # 确定启动命令，.py 用 python，.js 用 node
36
        command = "python" if is_python else "node"
37

38
        # 构造 MCP 所需的服务器参数StdioServerParameters，包含启动命令、脚本路径参数、环境变量（为 None 表示默认）
39
        server_params = StdioServerParameters(command=command, args=[server_script_path], env=None)
40

41
        # 启动 MCP 工具服务进程，stdio_client 创建 stdio 双向传输通信通道（子进程 stdin 写、stdout 读）
42
        stdio_transport = await self.exit_stack.enter_async_context(stdio_client(server_params))
43

44
        # 拆包通信通道：self.stdio(读流), self.write(写流)。作为 MCP 会话底层通信载体；
45
        self.stdio, self.write = stdio_transport
46

47
        # 创建 ClientSession（MCP 客户端会话对象，封装协议方法：list_tools/call_tool）
48
        self.session = await self.exit_stack.enter_async_context(ClientSession(self.stdio, self.write))
49

50
        # 初始化会话，执行 MCP 握手协议，建立正式通信；
51
        await self.session.initialize()
52

53
        # list_tools() 拉取服务端所有工具，缓存下来供后续规划使用。
54
        response = await self.session.list_tools()
55
        tools = response.tools
56
        print("\n已连接到服务器，支持以下工具:", [tool.name for tool in tools])
57

58
        # 总的来说，底层通信模型就是：
59
        # MCPClient(主程序) <==stdio管道==> server.py(工具服务进程)
60
        # write → 往子进程stdin发请求（调用工具、查询工具列表）
61
        # stdio → 读取子进程stdout返回结果
62
        # ClientSession 封装了这套管道的MCP协议报文序列化/反序列化，不用手动拼JSON
63

64

65
  # 工具链规划模块（自研核心逻辑，不属于原生 MCP）
66
  # 设计目的：减少多轮 LLM 调用开销，一次性规划完整任务流程，适合固定流水线任务（情感分析→生成报告→发邮件）。
67
  async def plan_tool_usage(self, query: str, tools: List[dict]) -> List[dict]:
68
        # 构造系统提示词 system_prompt。
69
        # 将所有可用工具组织为文本列表插入提示中，并明确指出工具名，
70
        # 限定返回格式是 JSON，防止其输出错误格式的数据。
71
        print("\n📤 提交给大模型的工具定义:")
72
        print(json.dumps(tools, ensure_ascii=False, indent=2))
73
        tool_list_text = "\n".join([
74
            f"- {tool['function']['name']}: {tool['function']['description']}"
75
            for tool in tools
76
        ])
77
        system_prompt = {
78
            "role": "system",
79
            "content": (
80
                "你是一个智能任务规划助手，用户会给出一句自然语言请求。\n"
81
                "你只能从以下工具中选择（严格使用工具名称）：\n"
82
                f"{tool_list_text}\n"
83
                "如果多个工具需要串联，后续步骤中可以使用 {{上一步工具名}} 占位。\n"
84
                "返回格式：JSON 数组，每个对象包含 name 和 arguments 字段。\n"
85
                "不要返回自然语言，不要使用未列出的工具名。"
86
            )
87
        }
88

89
        # 构造对话上下文并调用模型。
90
        # 将系统提示和用户的自然语言一起作为消息输入，并选用当前的模型。
91
        planning_messages = [
92
            system_prompt,
93
            {"role": "user", "content": query}
94
        ]
95

96
        response = self.client.chat.completions.create(
97
            model=self.model,
98
            messages=planning_messages,
99
            tools=tools,
100
            tool_choice="none"
101
        )
102

103
        # 提取出模型返回的 JSON 内容
104
        content = response.choices[0].message.content.strip()
105
        match = re.search(r"```(?:json)?\\s*([\s\S]+?)\\s*```", content)
106
        if match:
107
            json_text = match.group(1)
108
        else:
109
            json_text = content
110

111
        # 在解析 JSON 之后返回调用计划
112
        try:
113
            plan = json.loads(json_text)
114
            return plan if isinstance(plan, list) else []
115
        except Exception as e:
116
            print(f"❌ 工具调用链规划失败: {e}\n原始返回: {content}")
117
            return []
118

119
    async def cleanup(self):
120
        await self.exit_stack.aclose()
121

122

123
  # 业务主逻辑，整条流水线：
124
  # 用户输入 → 生成报告文件名 → LLM 规划工具链 → 循环执行工具 → 汇总结果给 LLM 生成最终回答 → 本地保存对话日志
125
    async def process_query(self, query: str) -> str:
126
        # 准备初始消息和获取工具列表
127
        messages = [{"role": "user", "content": query}]
128
        response = await self.session.list_tools()
129

130
        available_tools = [
131
            {
132
                "type": "function",
133
                "function": {
134
                    "name": tool.name,
135
                    "description": tool.description,
136
                    "input_schema": tool.inputSchema
137
                }
138
            } for tool in response.tools
139
        ]
140

141
        # 提取问题的关键词，对文件名进行生成。
142
        # 在接收到用户提问后就应该生成出最后输出的 md 文档的文件名，
143
        # 因为导出时若再生成文件名会导致部分组件无法识别该名称。
144
        keyword_match = re.search(r'(关于|分析|查询|搜索|查看)([^的\s，。、？\n]+)', query)
145
        keyword = keyword_match.group(2) if keyword_match else "分析对象"
146
        safe_keyword = re.sub(r'[\\/:*?"<>|]', '', keyword)[:20]
147
        timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
148
        md_filename = f"sentiment_{safe_keyword}_{timestamp}.md"
149
        md_path = os.path.join("./sentiment_reports", md_filename)
150

151
        # 更新查询，将文件名添加到原始查询中，使大模型在调用工具链时可以识别到该信息
152
        # 然后调用 plan_tool_usage 获取工具调用计划
153
        query = query.strip() + f" [md_filename={md_filename}] [md_path={md_path}]"
154
        messages = [{"role": "user", "content": query}]
155

156
        tool_plan = await self.plan_tool_usage(query, available_tools)
157

158
        tool_outputs = {}
159
        messages = [{"role": "user", "content": query}]
160

161
        # 依次执行工具调用，并收集结果
162
        for step in tool_plan:
163
            tool_name = step["name"]
164
            tool_args = step["arguments"]
165

166
            for key, val in tool_args.items():
167
                if isinstance(val, str) and val.startswith("{{") and val.endswith("}}"):
168
                    ref_key = val.strip("{} ")
169
                    resolved_val = tool_outputs.get(ref_key, val)
170
                    tool_args[key] = resolved_val
171

172
            # 注入统一的文件名或路径（用于分析和邮件）
173
            if tool_name == "analyze_sentiment" and "filename" not in tool_args:
174
                tool_args["filename"] = md_filename
175
            if (
176
                tool_name == "send_email_with_attachment"
177
                and "filename" not in tool_args
178
                and "attachment_path" not in tool_args
179
            ):
180
                tool_args["attachment_path"] = md_path
181

182
            result = await self.session.call_tool(tool_name, tool_args)
183

184
            tool_outputs[tool_name] = result.content[0].text
185
            messages.append({
186
                "role": "tool",
187
                "tool_call_id": tool_name,
188
                "content": result.content[0].text
189
            })
190

191
        # 调用大模型生成回复信息，并输出保存结果
192
        final_response = self.client.chat.completions.create(
193
            model=self.model,
194
            messages=messages
195
        )
196
        final_output = final_response.choices[0].message.content
197

198
        # 对辅助函数进行定义，目的是把文本清理成合法的文件名
199
        def clean_filename(text: str) -> str:
200
            text = text.strip()
201
            text = re.sub(r'[\\/:*?\"<>|]', '', text)
202
            return text[:50]
203

204
        # 使用清理函数处理用户查询，生成用于文件命名的前缀，并添加时间戳、设置输出目录
205
        # 最后构建出完整的文件路径用于保存记录
206
        safe_filename = clean_filename(query)
207
        timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
208
        filename = f"{safe_filename}_{timestamp}.txt"
209
        output_dir = "./llm_outputs"
210
        os.makedirs(output_dir, exist_ok=True)
211
        file_path = os.path.join(output_dir, filename)
212

213
        # 将对话内容写入 md 文档，其中包含用户的原始提问以及模型的最终回复结果
214
        with open(file_path, "w", encoding="utf-8") as f:
215
            f.write(f"🗣 用户提问：{query}\n\n")
216
            f.write(f"🤖 模型回复：\n{final_output}\n")
217

218
        print(f"📄 对话记录已保存为：{file_path}")
219

220
        return final_output
221

222
  # 交互入口
223
    async def chat_loop(self):
224
        # 初始化提示信息
225
        print("\n🤖 MCP 客户端已启动！输入 'quit' 退出")
226

227
        # 进入主循环中等待用户输入
228
        while True:
229
            try:
230
                query = input("\n你: ").strip()
231
                if query.lower() == 'quit':
232
                    break
233

234
                # 处理用户的提问，并返回结果
235
                response = await self.process_query(query)
236
                print(f"\n🤖 AI: {response}")
237

238
            except Exception as e:
239
                print(f"\n⚠️ 发生错误: {str(e)}")
240

241

242

243

244
async def main():
245
    server_script_path = "./server.py"
246
    client = MCPClient()
247
    try:
248
        await client.connect_to_server(server_script_path)
249
        await client.chat_loop()
250
    finally:
251
      # 资源释放：关闭 AsyncExitStack，自动关闭 MCP 会话、杀掉 server 子进程，防止资源泄漏；
252
        await client.cleanup()
253

254

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

1
your-skill-name/           # 技能文件夹
2
├── SKILL.md               # 必填: 核心指令文件, 告诉 AI 该怎么做
3
├── scripts/               # 可选: 可执行代码目录
4
│   └── process_data.py    # 示例: Python 处理脚本
5
├── references/             # 可选: 参考文档目录
6
│   └── api-guide.md       # 示例: API 接口说明文档
7
└── assets/                # 可选: 资源文件目录
8
    └── report-template.md # 示例: 报告生成模板

1
1. 用户发送请求
2
   ↓
3
2. Level 1 总是加载：读取 SKILL.md 头部 YAML 简介
4
   ↓
5
3. 判断用户请求是否匹配该技能
6
   ├─ 不匹配 → 走常规对话流程，结束
7
   └─ 匹配 → 进入 Level 2
8
   ↓
9
4. Level 2 触发时加载：读取 SKILL.md 完整指令、执行步骤
10
   ↓
11
5. AI 按照文档步骤开始执行任务
12
   ├─ 任务不需要外部文件资源 → 直接输出最终结果
13
   └─ 任务需要读取文档 / 运行脚本 → 进入 Level 3
14
   ↓
15
6. Level 3 引用时按需加载：读取外部资源目录
16
   ├─ references/ 参考文档：文档全文加载进上下文，再生成结果
17
   └─ scripts/ 可执行脚本：运行代码，仅将运行结果传入上下文，代码本身不占用上下文
18
   ↓
19
   汇总所有信息，输出最终结果

1
---
2
name: api-expert
3
description: FastAPI 开发的最佳实践。在构建、审查或调试FastAPI应用、REST API或Pydantic模型时使用。
4
metadata:
5
  pattern: tool-wrapper
6
  domain: fastapi
7
---
8

9
你是 FastAPI 开发领域的专家。请将以下规范应用于用户的代码或问题中。
10

11
## 核心规范
12

13
加载 'references/conventions.md'以获取完整的FastAPI最佳实践。
14

15
## 审查代码时
16
1. 加载规范参考文件
17
2. 逐条对照规范检查用户的代码
18
3. 对于每一处违反规范的地方，引用具体规则并提供修复建议。
19

20
## 编写代码时
21
1. 加载规范参考文件
22
2. 严格遵循每一项规范
23
3. 为所有函数签名添加类型注解
24
4. 使用 Annotated 风格进行依赖注入

1
---
2
name: report-generator
3
description: 生成结构化的Markdown格式技术报告。当用户要求编写、创建或起草报告、摘要或分析文档时使用。
4
metadata:
5
  pattern: generator
6
  output-format: markdown
7
---
8

9
你是一个技术报告生成器。请严格遵循以下步骤：
10

11
步骤 1: 加载 'references/style-guide.md'获取语气和排版规则。
12
步骤 2: 加载 'assets/report-template.md'获取所需的输出结构。
13
步骤 3: 向用户询问填充模板所需的任何缺少信息：
14
- 主题或内容
15
- 关键发现或数据点
16
- 目标受众（技术人员、管理层、普通受众）
17

18
步骤 4: 遵循样式指南的规则填充模板。模板中的每个部分都必须出现在最终的输出中。
19

20
步骤 5: 将完成的报告作为单个 Markdown文档返回。

1
---
2
name: code-reviewer
3
description: 审查 Python 代码的质量、风格和常见Bug。当用户提交代码以供审查、要求获取代码反馈或希望进行代码审计时使用。
4
metadata:
5
  pattern: reviewer
6
  severity-levels: error,warning，info
7
---
8

9
你是一个 Python 代码审查员。请严格遵循以下审查协议：
10

11
步骤 1: 加载 'references/review-checkkist.md'获取完整的审查标准。
12

13
步骤 2: 仔细阅读用户的代码。在进行点评之前，先理解其意图。
14

15
步骤 3: 将清单中的每条规则应用于代码。对于发现的每一个违规之处：
16
- 记录行号（或大致位置）
17
- 对严重程度进行分类: error（必须修复）、warning(建议修复)、info(供参考)
18
- 解释“为什么”这是一个问题, 而不仅仅是指出”哪里“错了
19
- 提供具体的修复建议及修改后的代码
20

21
步骤 4: 生成包含以下部分的结构化审查报告：
22
- **总结**: 代码的功能说明，整体质量评估
23
- **发现的问题**: 按严重程度分组（error优先，其次是warning，最后是info）
24
- **评分**: 1-10 分，并附带简短理由

1
---
2
name: project-planner
3
description: 规划新软件项目，在生成计划之前通过结构化的问题收集需求。当用户说“我想构建”、“帮我做计划”、“设计系统”或“开始新项目”时使用。
4
metadata:
5
  pattern: inversion
6
  interaction: multi-turn
7
---
8

9
你正在进行一次结构化的需求访谈。在所有阶段完成之前，绝对不要开始构建或设计。
10

11
## 阶段 1 – 问题发现（一次只问一个问题，等待每个回答）
12
按顺序提出以下问题。不要跳过任何一个
13
- Q1: “这个项目为用户解决了什么问题？”
14
- Q2: “主要用户是谁？他们的技术水平如何？”
15
- Q3: “预期的规模有多大？（每日用户量、数据量、请求频率）”
16

17
## 阶段 2 – 技术约束（仅在阶段 1 被被完全解答后进行)
18
- Q4: “你打算使用什么部署环境？”
19
- Q5: “你对技术栈有什么要求或偏好吗？”
20
- Q6: “有哪些不可妥协的要求？（延迟、正常运行时间、合规性、预算) ”
21

22
## 阶段 3 – 总结（仅在所有问题都被解答后进行)
23
1. 加载 'assets/plan-template.md'获取输出格式
24
2. 使用收集到的需求填充模板的每个部分
25
3. 向用户展示完整的计划
26
4. 询问: “这个计划准确地涵盖了你的需求吗？你希望修改哪些地方？
27
5. 根据反馈不断迭代, 直到用户确认为止
28
```

1
---
2
name: doc-pipeline
3
description: 通过多步流水线从 Python 源代码生成 API 文档。当用户要求为模块编写文档、生成API文档或从代码创建文档时使用。
4
metadata:
5
  pattern: pipeline
6
  steps: "4"
7
---
8

9
你正在运行一个文档生成流水线。请严格请严格按顺序执行每个步骤。绝对不要跳过任何步骤，也不要在某个步骤失败时继续前进。
10

11
## 步骤 1 – 解析与盘点
12
分析用户的 Python 代码，提取所有公共的类、函数和常量。将盘点清单作为检查列表展示给用户，并询问：“这是您希望编写文档的完整公共API吗？”
13

14
## 步骤 2 – 生成文档字符串（Docstring）
15
对于每一个缺少文档字符串的函数:
16
- 加载 'references/docstring-style.md' 获取所需的格式规范
17
- 严格遵循样式指南生成文档字符串
18
- 向用户展示每个生成的文档字符串以供审核批准
19
**在用户确认之前，绝对不要进入步骤 3。**
20

21
## 步骤 3 – 组装文档
22
加载 'assets/api-doc-template.md'获取输出结构。将所有的类、函数和文档字符串编译整合进一个单一的API参考文档中。
23

24
## 步骤 4 – 质量检查
25
根据 'references/quality-checklis'进行复核:
26
- 确保每一个公共符号都已记录
27
- 确保每一个参数都包含类型和描述
28
- 确保每个函数至少提供一个使用示例
29
汇报检查结果。在展示最终文档之前，必须先修复发现的所有问题。

1
---
2
name: systematic-debugging
3
description: "4-phase root cause debugging: understand bugs before fixing."
4
version: 1.1.0
5
author: Hermes Agent (adapted from obra/superpowers)
6
license: MIT
7
metadata:
8
  hermes:
9
    tags: [debugging, troubleshooting, problem-solving, root-cause, investigation]
10
    related_skills: [test-driven-development, writing-plans, subagent-driven-development]
11
---