從源碼分析 MCP 的實現和使用
MCP（Model Context Protocol）是最近非常熱門的詞，本文從實現和使用兩個方面介紹 MCP 的內容。
MCP 是什麼？
官方的定義如下：
MCP is an open protocol that standardizes how applications provide context to LLMs. Think of MCP like a USB-C port for AI applications. Just as USB-C provides a standardized way to connect your devices to various peripherals and accessories, MCP provides a standardized way to connect AI models to different data sources and tools.
從實現來看，MCP 定義了一套通用的 API 用於 MCP 客戶端和服務器通信，MCP 開發者只需要基於 MCP 提供的 Server 註冊自己的工具、Prompt，就可以通過 MCP 提供的 Client 獲取並調用這些能力，官網給的圖如下。
MCP 提供的客戶端和服務器實現了通用的能力，比如如何通信，如何獲取工具等，具體有哪些能力（比如由哪裏工具）是由 Server 開發者定義的，我們可以自己開發工具，也可以使用第三方開源的。但如何使用這些能力是用使用者決定的，比如我們開發一個基於 Redis 的 MCP Server，那麼我們就可以通過 MCP Client 從這個 Server 從 Redis 裏獲取信息，但是如何使用這些信息是自己定義的，一般來說就是輸入給大模型。
瞭解了 MCP Server 和 Client 架構後，接着看看 MCP 在實際場景中的使用架構。下圖（來自《 MCP 是什麼，現狀和未來？）描述了 MCP 在用戶和大模型通信過程中的位置和作用。
上圖的流程大致如下。
-
用戶通過 MCP Client 從 MCP Server 獲取可用的能力，比如工具、Prompt。
-
把工具和用戶輸入傳入大模型，大模型返回時會告訴用戶可以使用哪些工具去獲取更多信息。
-
用戶再次通過 MCP Client 請求 MCP Server，讓 MCP Server 執行對應的工具獲取數據。
-
用戶從 MCP Server 獲取數據後，再次訪問大模型，大模型就可以利用工具的信息更好地回解決用戶的問題。
MCP 的使用例子
瞭解了 MCP 的一些基礎概念後，來看一下 MCP 使用的例子。
import { Server } from "@modelcontextprotocol/sdk/server/index.js";
import { SSEServerTransport } from "@modelcontextprotocol/sdk/server/sse.js";
import express from 'express';
import { z } from 'zod';
import { CallToolRequestSchema, ListToolsRequestSchema, } from "@modelcontextprotocol/sdk/types.js";
const { Request, Response } = express;
const GetArgumentsSchema = z.object({
key: z.string(),
});
// 模擬 redis
const memory = {"hello": "world"};
// 創建 MCP Server
const server = new Server({
name: "example-server",
version: "1.0.0",
},{
capabilities: {
tools: {},
}
});
// 註冊獲取工具 API 路由
server.setRequestHandler(ListToolsRequestSchema, async () => {
return {
tools: [
{
name: "get",
description: "Get value by key from kv",
inputSchema: {
type: "object",
properties: {
key: {
type: "string",
description: "",
},
},
required: ["key"],
},
},
],
};
});
// 註冊執行工具 API 路由
server.setRequestHandler(CallToolRequestSchema, async (request) => {
console.log(request.params)
const { name, arguments: args } = request.params;
const { key } = GetArgumentsSchema.parse(args);
return {
content: [
{
type: "text",
text: `${memory[key]}`,
},
],
};
});
// 啓動 HTTP 服務器
const app = express();
const transports: {[sessionId: string]: SSEServerTransport} = {};
// SSE 路由
app.get("/sse", async (_: Request, res: Response) => {
// 創建一個 SSE 通道，meesages 爲後續通信的路由
const transport = new SSEServerTransport('/messages', res);
transports[transport.sessionId] = transport;
res.on("close", () => {
delete transports[transport.sessionId];
});
await server.connect(transport);
});
// 通信路由
app.post("/messages", async (req: Request, res: Response) => {
const sessionId = req.query.sessionId as string;
const transport = transports[sessionId];
if (transport) {
// 處理請求，處理完畢後通過上面的 sse res 進行響應
await transport.handlePostMessage(req, res);
} else {
res.status(400).send('No transport found for sessionId');
}
});
app.listen(3001);MCP Server 並不是一個傳統的 HTTP Server，它只是負責處理來自客戶端的請求，並不處理 HTTP 協議本身，其中，Server 的 setRequestHandler API 用於註冊路由和處理函數，第一個 setRequestHandler 註冊了獲取工具列表 API，第二個 setRequestHandler 註冊了執行工具的 API，客戶端一般先獲取工具列表，然後再調 API 執行具體的工具。創建 MCP Server 後，還需要創建一個通信管道用戶接收客戶端的請求，這裏是 SSE 通道，SSE 會接收到來自 HTTP 服務器的請求，然後傳遞給 MCP Server，MCP Server 再通過通道把處理結果返回給客戶端。
接着看客戶端的例子。
import { Client } from "@modelcontextprotocol/sdk/client/index.js";
import { SSEClientTransport } from "@modelcontextprotocol/sdk/client/sse.js";
async function runClient() {
const client = new Client(
{
name: "mcp-typescript test client",
version: "0.1.0",
},
);
const clientTransport = new SSEClientTransport(new URL("http://localhost:3001/sse"));
await client.connect(clientTransport);
const tools = await client.listTools()
console.log(tools)
const resp = await client.callTool({
name: "get",
arguments: {
key: "hello"
}
})
console.log(resp);
await client.close();
}
runClient();執行上面代碼會首先從 MCP 客戶端獲取工具列表，然後執行 get 命令獲取數據，客戶端和服務端的架構類似，就不再介紹。
MCP 的實現
Transport
MCP 客戶端和服務器需要通信來完成數據的交互，而通信就需要一個通道，所以先看一下通道的實現。MCP 支持通過本地、SSE、等方式來通信，通過 Transport 實現了抽象。
interface Transport {
// 初始化通道
start(): Promise<void>;
// 發送信息或響應
send(message: JSONRPCMessage): Promise<void>;
close(): Promise<void>;
onclose?: () => void;
onerror?: (error: Error) => void;
// 處理收到的請求或響應
onmessage?: (message: JSONRPCMessage) => void;
// 連接對應的回話 ID
sessionId?: string;
}Transport 可以用於客戶端或服務端，只要實現了上面的方法就行。
我們來看一個基於內存的通信管道的實現，類似 Unix 管道。
class InMemoryTransport implements Transport {
private _otherTransport?: InMemoryTransport;
private _messageQueue: JSONRPCMessage[] = [];
onclose?: () => void;
onerror?: (error: Error) => void;
onmessage?: (message: JSONRPCMessage) => void;
sessionId?: string;
static createLinkedPair(): [InMemoryTransport, InMemoryTransport] {
// 創建兩個管道，互相關聯
const clientTransport = new InMemoryTransport();
const serverTransport = new InMemoryTransport();
clientTransport._otherTransport = serverTransport;
serverTransport._otherTransport = clientTransport;
return [clientTransport, serverTransport];
}
// 初始化管道，如果之前已經存在數據，則消費
async start(): Promise<void> {
while (this._messageQueue.length > 0) {
const message = this._messageQueue.shift();
if (message) {
this.onmessage?.(message);
}
}
}
async close(): Promise<void> {
const other = this._otherTransport;
this._otherTransport = undefined;
await other?.close();
this.onclose?.();
}
// 發送數據
async send(message: JSONRPCMessage): Promise<void> {
if (!this._otherTransport) {
throw new Error("Not connected");
}
// 如果還沒有消費者，則先緩存
if (this._otherTransport.onmessage) {
this._otherTransport.onmessage(message);
} else {
this._otherTransport._messageQueue.push(message);
}
}
}在實際場景中，我們一般使用 SSE 或 Websocket 來實現通信，下面看一下 Server SSE Transport 的實現。
class SSEServerTransport implements Transport {
private _sseResponse?: ServerResponse;
private _sessionId: string;
onclose?: () => void;
onerror?: (error: Error) => void;
// 上層設置
onmessage?: (message: JSONRPCMessage) => void;
// 初始化時記錄 SSE 連接對應的響應對象，以及用於通信的 endpoint 路由
constructor(
private _endpoint: string,
private res: ServerResponse,
) {
this._sessionId = randomUUID();
}
async start(): Promise<void> {
// 返回 text/event-stream 類型的響應頭，表示這是一個 SSE 連接，後續可以基於這個連接進行數據推送
this.res.writeHead(200, {
"Content-Type": "text/event-stream",
"Cache-Control": "no-cache",
Connection: "keep-alive",
});
// 返回 endpoint 信息，作爲 POST 請求的 url 路徑
this.res.write(
`event: endpoint\ndata: ${encodeURI(this._endpoint)}?sessionId=${this._sessionId}\n\n`,
);
// 記錄 SSE 連接對應的響應對象
this._sseResponse = this.res;
}
// 處理 POST 請求，解析請求體，將請求體並通過 SSE 響應對象把請求結果發送給客戶端
async handlePostMessage(
req: IncomingMessage,
res: ServerResponse,
parsedBody?: unknown,
): Promise<void> {
// 請求前置處理
let body: string | unknown;
try {
const ct = contentType.parse(req.headers["content-type"] ?? "");
if (ct.type !== "application/json") {
throw new Error(`Unsupported content-type: ${ct}`);
}
body = parsedBody ?? await getRawBody(req, {
limit: MAXIMUM_MESSAGE_SIZE,
encoding: ct.parameters.charset ?? "utf-8",
});
} catch (error) {
res.writeHead(400).end(String(error));
this.onerror?.(error as Error);
return;
}
try {
// 具體的處理邏輯
await this.handleMessage(typeof body === 'string' ? JSON.parse(body) : body);
} catch {
res.writeHead(400).end(`Invalid message: ${body}`);
return;
}
// 先回復 202 Accepted，表示請求已經被接受，後續處理邏輯由上層處理
res.writeHead(202).end("Accepted");
}
async handleMessage(message: unknown): Promise<void> {
let parsedMessage: JSONRPCMessage;
// 解析請求，通知上層處理
try {
parsedMessage = JSONRPCMessageSchema.parse(message);
} catch (error) {
this.onerror?.(error as Error);
throw error;
}
this.onmessage?.(parsedMessage);
}
async send(message: JSONRPCMessage): Promise<void> {
this._sseResponse.write(
`event: message\ndata: ${JSON.stringify(message)}\n\n`,
);
}
get sessionId(): string {
return this._sessionId;
}
}SSE 的處理過程如下。
-
客戶端發起 SSE 請求，服務器收到 SSE 請求時，記錄對應的響應對象和設置通信路由，返回通信路由給客戶端。
-
客戶端收到 SSE 通信路徑後，後續通過 HTTP POST 請求到服務端返回的路由。
-
服務端收到 HTTP Post 請求時，先返回 202 表示請求已被成功接收，然後通知上層，上冊處理完畢後通過第一步保存的 SSE 響應對象推送處理結果給客戶端。
接着看一下 Client 的 SSE Transport 的實現。
export class SSEClientTransport implements Transport {
private _eventSource?: EventSource;
private _endpoint?: URL;
private _abortController?: AbortController;
private _url: URL;
private _eventSourceInit?: EventSourceInit;
private _requestInit?: RequestInit;
// 上層定義
onmessage?: (message: JSONRPCMessage) => void;
constructor(
url: URL,
opts?: SSEClientTransportOptions,
) {
this._url = url;
this._eventSourceInit = opts?.eventSourceInit;
this._requestInit = opts?.requestInit;
// 身份驗證提供者
this._authProvider = opts?.authProvider;
}
async start() {
return await this._startOrAuth();
}
private _startOrAuth(): Promise<void> {
return new Promise((resolve, reject) => {
// SSE 客戶端
this._eventSource = new EventSource(
this._url.href,
this._eventSourceInit ?? {
// 自定義請求實現
fetch: (url, init) => this._commonHeaders().then((headers) => fetch(url, {
...init,
headers: {
...headers,
Accept: "text/event-stream"
}
})),
},
);
// 接收服務器的 endpoint，用於後續通信
this._eventSource.addEventListener("endpoint", (event: Event) => {
const messageEvent = event as MessageEvent;
this._endpoint = new URL(messageEvent.data, this._url);
resolve();
});
// 接收服務端消息
this._eventSource.onmessage = (event: Event) => {
const messageEvent = event as MessageEvent;
let message: JSONRPCMessage;
try {
message = JSONRPCMessageSchema.parse(JSON.parse(messageEvent.data));
} catch (error) {
this.onerror?.(error as Error);
return;
}
this.onmessage?.(message);
};
});
}
// 發送消息給客戶端
async send(message: JSONRPCMessage): Promise<void> {
const headers = new Headers({ ...this._requestInit?.headers });
headers.set("content-type", "application/json");
const init = {
...this._requestInit,
method: "POST",
headers,
body: JSON.stringify(message),
signal: this._abortController?.signal,
};
const response = await fetch(this._endpoint, init);
}
}Server
瞭解了 Transport 後，繼續看 Server 是如何處理來自 Transport 的請求的。
class Server extends Protocol {
private _clientCapabilities?: ClientCapabilities;
private _clientVersion?: Implementation;
private _capabilities: ServerCapabilities;
private _instructions?: string;
constructor(
private _serverInfo: Implementation,
options?: ServerOptions,
) {
super(options);
// 支持的能力
this._capabilities = options?.capabilities ?? {};
this._instructions = options?.instructions;
// 註冊初始化請求路由和處理函數，該請求用戶獲取服務端的一些元信息，比如支持的能力
this.setRequestHandler(InitializeRequestSchema, (request) =>
this._oninitialize(request),
);
}
private async _oninitialize(
request: InitializeRequest,
): Promise<InitializeResult> {
const requestedVersion = request.params.protocolVersion;
this._clientCapabilities = request.params.capabilities;
this._clientVersion = request.params.clientInfo;
return {
protocolVersion: SUPPORTED_PROTOCOL_VERSIONS.includes(requestedVersion)
? requestedVersion
: LATEST_PROTOCOL_VERSION,
capabilities: this.getCapabilities(),
serverInfo: this._serverInfo,
...(this._instructions && { instructions: this._instructions }),
};
}
// 註冊 Server 支持的能力，比如工具、Prompt
public registerCapabilities(capabilities: ServerCapabilities): void {
this._capabilities = mergeCapabilities(this._capabilities, capabilities);
}
}Server 繼承了 Protocol，本身處理了 InitializeRequestSchema 請求，用於返回服務端的元信息，接着看 Protocol。
Protocol
export abstract class Protocol {
private _transport?: Transport;
private _requestMessageId = 0;
private _requestHandlers = new Map();
private _responseHandlers = new Map();
// 初始化實例，保存 transport，後續用 transport 和對象通信
async connect(transport: Transport): Promise<void> {
this._transport = transport;
this._transport.onmessage = (message) => {
// 根據 message 的類型，分別處理請求、響應、通知，因爲該類會被 Clien 和 Server 繼承
if (!("method" in message)) {
this._onresponse(message);
} else if ("id" in message) {
this._onrequest(message);
} else {
this._onnotification(message);
}
};
await this._transport.start();
}
private _onrequest(request: JSONRPCRequest): void {
const handler = this._requestHandlers.get(request.method) ?? this.fallbackRequestHandler;
const extra = {
sessionId: this._transport?.sessionId,
};
Promise.resolve()
.then(() => handler(request, extra)) // 處理請求
.then(
(result) => {
// 發送處理請求的結果
return this._transport?.send({
result,
jsonrpc: "2.0",
id: request.id,
});
})
}
private _onresponse(response: JSONRPCResponse | JSONRPCError): void {
const messageId = Number(response.id);
const handler = this._responseHandlers.get(messageId);
this._responseHandlers.delete(messageId);
handler(response);
}
request<T extends ZodType<object>>(
request: SendRequestT,
resultSchema: T,
options?: RequestOptions,
): Promise<z.infer<T>> {
return new Promise((resolve, reject) => {
// 請求 ID 遞增
const messageId = this._requestMessageId++;
const jsonrpcRequest: JSONRPCRequest = {
...request,
jsonrpc: "2.0",
id: messageId,
};
// 設置請求和處理函數的映射，在 _onresponse 處理響應
this._responseHandlers.set(messageId, (response) => {
try {
const result = resultSchema.parse(response.result);
resolve(result);
} catch (error) {
reject(error);
}
});
// 發送請求
this._transport.send(jsonrpcRequest);
});
}
// 註冊路由和處理函數
setRequestHandler<
T extends ZodObject<{
method: ZodLiteral<string>;
}>,
>(
requestSchema: T,
handler: (
request: z.infer<T>,
extra: RequestHandlerExtra,
) => SendResultT | Promise<SendResultT>,
): void {
const method = requestSchema.shape.method.value;
// 存入 map
this._requestHandlers.set(method, (request, extra) =>
Promise.resolve(handler(requestSchema.parse(request), extra)),
);
}
}Protocol 基於 Transport 實現了對請求和響應的封裝，具體請求和響應的處理由上層的 Client 或 Server 執行。
MCP Server
通過前面的分析可以知道，已經實現了數據通信、路由註冊和處理，請求和響應的封裝，接着看 MCP Server 的實現，MCP Server 在之前的基礎上實現了一系列對外的 API 和提供註冊工具、Prompt 等能力。
export class McpServer {
public readonly server: Server;
private _registeredTools = {};
constructor(serverInfo: Implementation, options?: ServerOptions) {
private _registeredTools: { [name: string]: Tool } = {};
}
async connect(transport: Transport): Promise<void> {
return await this.server.connect(transport);
}
private setToolRequestHandlers() {
// 表示 Server 支持工具
this.server.registerCapabilities({
tools: {},
});
// 註冊獲取工具列表路由
this.server.setRequestHandler(
ListToolsRequestSchema,
(): ListToolsResult => ({
tools: Object.entries(this._registeredTools).map(
([name, tool]): Tool => {
return {
name,
description: tool.description,
inputSchema: tool.inputSchema
? (zodToJsonSchema(tool.inputSchema, {
strictUnions: true,
}) as Tool["inputSchema"])
: EMPTY_OBJECT_JSON_SCHEMA,
};
},
),
}),
);
// 註冊調用工具路由
this.server.setRequestHandler(
CallToolRequestSchema,
async (request, extra): Promise<CallToolResult> => {
// 獲取對應的工具
const tool = this._registeredTools[request.params.name];
const parseResult = await tool.inputSchema.safeParseAsync(
request.params.arguments,
);
// 解析參數
const args = parseResult.data;
const cb = tool.callback as ToolCallback<ZodRawShape>;
return await Promise.resolve(cb(args, extra));
},
);
}
// 註冊工具，name爲工具名，description爲工具描述，paramsSchema爲參數的zod schema，cb爲工具的回調函數
tool(name: string, ...rest: unknown[]): void {
if (this._registeredTools[name]) {
throw new Error(`Tool ${name} is already registered`);
}
let description;
if (typeof rest[0] === "string") {
description = rest.shift() as string;
}
let paramsSchema;
if (rest.length > 1) {
paramsSchema = rest.shift() as ZodRawShape;
}
const cb = rest[0];
// 記錄工具信息
this._registeredTools[name] = {
description,
inputSchema:
paramsSchema === undefined ? undefined : z.object(paramsSchema),
callback: cb,
};
// 給 MCP Server 註冊路由，這樣就可以處理客戶端的工具相關請求
this.setToolRequestHandlers();
}
}MCP Server 主要提供了一些 API 方便用戶註冊工具、Prompt 等，如果註冊了相關的能力，那麼 MCP Server 就會註冊對應的路由，這樣客戶端就可以訪問這些路由獲取相關能力。
Client
Client 繼承了 Protocol 並通過 Transport 實現和 MCP Server 的通信 。
export class Client extends Protocol {
private _serverCapabilities?: ServerCapabilities;
private _serverVersion?: Implementation;
private _capabilities: ClientCapabilities;
private _instructions?: string;
constructor(
private _clientInfo: Implementation,
options?: ClientOptions,
) {
super(options);
// 客戶端支持的能力
this._capabilities = options?.capabilities ?? {};
}
public registerCapabilities(capabilities: ClientCapabilities): void {
this._capabilities = mergeCapabilities(this._capabilities, capabilities);
}
override async connect(transport: Transport): Promise<void> {
// 建立和服務器的連接和身份驗證，由 Transport 實現提供
await super.connect(transport);
try {
// 發送初始化請求獲取服務器元信息，比如支持的能力
const result = await this.request(
{
method: "initialize",
params: {
protocolVersion: LATEST_PROTOCOL_VERSION,
capabilities: this._capabilities,
clientInfo: this._clientInfo,
},
},
InitializeResultSchema,
);
if (result === undefined) {
throw new Error(`Server sent invalid initialize result: ${result}`);
}
if (!SUPPORTED_PROTOCOL_VERSIONS.includes(result.protocolVersion)) {
throw new Error(
`Server's protocol version is not supported: ${result.protocolVersion}`,
);
}
this._serverCapabilities = result.capabilities;
this._serverVersion = result.serverInfo;
this._instructions = result.instructions;
await this.notification({
method: "notifications/initialized",
});
} catch (error) {
// Disconnect if initialization fails.
void this.close();
throw error;
}
}
// 調用工具
async callTool(
params: CallToolRequest["params"],
resultSchema,
options?: RequestOptions,
) {
return this.request(
{ method: "tools/call", params },
resultSchema,
options,
);
}
// 獲取工具列表
async listTools(
params?: ListToolsRequest["params"],
options?: RequestOptions,
) {
return this.request(
{ method: "tools/list", params },
ListToolsResultSchema,
options,
);
}
}Client 封裝了 MCP 協議定義的一系列 API，比如獲取工具、Prompt 列表，調用工具等。下面是 MCP 客戶端和服務器一次請求的通信過程。
MCP 在 OpenManus 中使用
下面以 OpenManus 爲例，看看 MCP 是如何和大模型結合使用的。下面是 run_mcp.py 的代碼。
async def run_mcp() -> None:
"""Main entry point for the MCP runner."""
args = parse_args()
runner = MCPRunner()
await runner.initialize(args.connection, args.server_url)
await runner.run_default()
if __name__ == "__main__":
asyncio.run(run_mcp())上面的代碼中創建了一個 MCPRunner 對象，然後執行它的 initialize 和 run_default 方法，看看 MCPRunner 的實現。
class MCPRunner:
def __init__(self):
self.agent = MCPAgent()
async def initialize(
self,
connection_type: str,
server_url: str | None = None,
) -> None:
// 建立和 MCP 服務器的連接
await self.agent.initialize(connection_type="sse", server_url=server_url)
async def run_default(self) -> None:
# 提示用戶輸入問題
prompt = input("Enter your prompt: ")
# 開始處理用戶輸入的問題
await self.agent.run(prompt)MCPRunner 是對 MCPAgent 封裝，並最終調了 MCPAgent 的 initialize 和 run 方法。接着看 MCPAgent 的實現。
class MCPAgent(ToolCallAgent):
mcp_clients: MCPClients = Field(default_factory=MCPClients)
async def initialize(
self,
connection_type: Optional[str] = None,
server_url: Optional[str] = None,
command: Optional[str] = None,
args: Optional[List[str]] = None,
) -> None:
# 初始化和 MCP 服務器的連接，並獲取工具列表
await self.mcp_clients.connect_sse(server_url=server_url)
# 保存工具相關的實例到 available_tools
self.available_tools = self.mcp_clients
async def run(self, request: Optional[str] = None) -> str:
result = await super().run(request)
return resultMCPAgent 通過 MCPClients 建立和 MCP Server 的連接，然後獲取工具列表。看看 MCP Client 的實現。
class MCPClients(ToolCollection):
session: Optional[ClientSession] = None
exit_stack: AsyncExitStack = None
description: str = "MCP client tools for server interaction"
def __init__(self):
super().__init__() # Initialize with empty tools list
self.name = "mcp" # Keep name for backward compatibility
self.exit_stack = AsyncExitStack()
async def connect_sse(self, server_url: str) -> None:
streams_context = sse_client(url=server_url)
streams = await self.exit_stack.enter_async_context(streams_context)
self.session = await self.exit_stack.enter_async_context(
ClientSession(*streams)
)
await self._initialize_and_list_tools()
# 從 MCP Server 獲取工具列表
async def _initialize_and_list_tools(self) -> None:
await self.session.initialize()
# 從 MCP Server 獲取工具列表
response = await self.session.list_tools()
# 存起來
self.tools = tuple()
self.tool_map = {}
# Create proper tool objects for each server tool
for tool in response.tools:
server_tool = MCPClientTool(
name=tool.name,
description=tool.description,
parameters=tool.inputSchema,
session=self.session,
)
self.tool_map[tool.name] = server_tool
self.tools = tuple(self.tool_map.values())獲取工具列表後，就執行 MCPAgent 的 run 啓動。
async def run(self, request: Optional[str] = None) -> str:
if self.state != AgentState.IDLE:
raise RuntimeError(f"Cannot run agent from state: {self.state}")
if request:
self.update_memory("user", request)
results: List[str] = []
# 循環調大模型
async with self.state_context(AgentState.RUNNING):
while (
self.current_step < self.max_steps and self.state != AgentState.FINISHED
):
self.current_step += 1
logger.info(f"Executing step {self.current_step}/{self.max_steps}")
# 不斷調 step，子類實現
step_result = await self.step()
results.append(f"Step {self.current_step}: {step_result}")
if self.current_step >= self.max_steps:
self.current_step = 0
self.state = AgentState.IDLE
results.append(f"Terminated: Reached max steps ({self.max_steps})")
await SANDBOX_CLIENT.cleanup()
return "\n".join(results) if results else "No steps executed"run 不斷調 step 方法，step 方法由子類實現。通過繼承鏈可以看到 MCPAgent 繼承 ToolCallAgent，ToolCallAgent 繼承 ReActAgent，ReActAgent 實現了 step。
async def step(self) -> str:
"""Execute a single step: think and act."""
should_act = await self.think()
if not should_act:
return "Thinking complete - no action needed"
return await self.act()step 中首先調 think 然後再調 act。這兩個方法由 ToolCallAgent 實現。
async def think(self) -> bool:
try:
# 傳入工具調大模型，詢問需要執行的工具
response = await self.llm.ask_tool(
messages=self.messages,
# 傳入可用的工具
tools=self.available_tools.to_params(),
tool_choice=self.tool_choices,
)
self.tool_calls = tool_calls = (
response.tool_calls if response and response.tool_calls else []
)
async def act(self) -> str:
results = []
# 執行模型返回的工具列表
for command in self.tool_calls:
result = await self.execute_tool(command)
results.append(result)
return "\n\n".join(results)
# 執行某個工具
async def execute_tool(self, command: ToolCall) -> str:
name = command.function.name
if name not in self.available_tools.tool_map:
return f"Error: Unknown tool '{name}'"
try:
# 解析大模型提取的參數
args = json.loads(command.function.arguments or "{}")
# 執行工具
result = await self.available_tools.execute(name=name, tool_input=args)
# Format result for display (standard case)
observation = (
f"Observed output of cmd `{name}` executed:\n{str(result)}"
if result
else f"Cmd `{name}` completed with no output"
)
return observation這裏以 OpenManus 提供的 Bash 工具爲例。
class _BashSession:
command: str = "/bin/bash"
async def start(self):
# 創建一個 bash 進程
self._process = await asyncio.create_subprocess_shell(
self.command,
preexec_fn=os.setsid,
shell=True,
bufsize=0,
stdin=asyncio.subprocess.PIPE,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
)
async def run(self, command: str):
# 讓 bash 執行命令
self._process.stdin.write(
command.encode() + f"; echo '{self._sentinel}'\n".encode()
)
await self._process.stdin.drain()
# 獲取輸出
class Bash(BaseTool):
"""A tool for executing bash commands"""
name: str = "bash"
description: str = _BASH_DESCRIPTION
parameters: dict = {
"type": "object",
"properties": {
"command": {
"type": "string",
"description": "The bash command to execute. Can be empty to view additional logs when previous exit code is `-1`. Can be `ctrl+c` to interrupt the currently running process.",
},
},
"required": ["command"],
}
_session: Optional[_BashSession] = None
async def execute(
self, command: str | None = None, restart: bool = False, **kwargs
) -> CLIResult:
if restart:
if self._session:
self._session.stop()
self._session = _BashSession()
await self._session.start()
return CLIResult(system="tool has been restarted.")
if self._session is None:
self._session = _BashSession()
await self._session.start()
if command is not None:
return await self._session.run(command)
raise ToolError("no command provided.")Bash 工具或創建一個子進程並執行大模型從用戶輸入中提取的命令。從代碼來看 OpenManus 只是收集並展示工具的輸出，而沒有再次把工具的輸出傳給模型進行下一步的查詢。
參考資料：
https://github.com/mannaandpoem/OpenManus/tree/main
https://github.com/modelcontextprotocol/typescript-sdk
https://github.com/modelcontextprotocol/servers
Introduction - Model Context Protocol
https://onevcat.com/2025/02/mcp/
本文由 Readfog 進行 AMP 轉碼，版權歸原作者所有。
來源：https://mp.weixin.qq.com/s/2YitPqj6UKQJio3ArVChew