AI Agent 专题 - 自建一个MCP服务
或许有一天，你也要搭建一个自己的MCP服务。
Background
作为上一篇文章个人代理的延续，今天我们来build 一个自定义的MCP 服务。
为什么要自建？
当我有某个基础服务，需要给多个agent 提供能力时，或者有些服务是业务垂直的，仅仅在我的公司里有许多用处，我就需要一个MCP服务来统一管理这些能力。
一个MCP服务需要什么？
Must have:
- 定义tools上下文的能力：能够定义符合Model Context Protocol规范的tools
- 具备tools的执行能力：能够与tools集成
- 会话状态管理的能力
Better to have:
- 支持多种不同的连接方式，比如websocket, http, grpc等
- 支持不同的认证方式，比如token, oauth等
MCP服务不需要什么？
MCP 服务是一个很简单的执行层，它并不需要任何llm的能力。
咸盐少许，我们开始
这个例子里，我们还是使用ADK来搭建我们都MCP 服务。
实现一个MCP 需要实现list tools 和call tools的方法
在adk里，就是通过这两个注解来告诉框架，调用哪个方法。
- @app.list_tools()
- @app.call_tool()
@app.list_tools()
list tools 能够允许你列出，你希望agent知道的tools。例子里我们用现成的tool，来自googl的tool库。
https://google.github.io/adk-docs/api-reference/python/google-adk.html#module-google.adk.tools
load_web_page tool：Fetches the content in the url and returns the text in it.
根据文档描述，load_web_page tool是以文字方式获取网页内容的一个功能。
1
2
3
4
5
6
7
8
9
10
11
adk_tool_to_expose = FunctionTool(load_web_page)
# Implement the MCP server's handler to list available tools
@app.list_tools()
async def list_mcp_tools() -> list[mcp_types.Tool]:
"""MCP handler to list tools this server exposes."""
print("MCP Server: Received list_tools request.")
# Convert the ADK tool's definition to the MCP Tool schema format
mcp_tool_schema = adk_to_mcp_tool_type(adk_tool_to_expose)
print(f"MCP Server: Advertising tool: {mcp_tool_schema.name}")
return [mcp_tool_schema]
@app.call_tool()
这是执行的实现
1
2
3
4
5
6
# Implement the MCP server's handler to execute a tool call
@app.call_tool()
async def call_mcp_tool(
name: str, arguments: dict
) -> list[mcp_types.Content]:
# ...
启动MCP 服务
MCP本质是个服务， ADK提供了多这个服务的种连接方式
- sse
- streamable_http
- websocket.py 例子里我们还是使用sse 来创建连接。sse 需要实现的是
Route("/sse", endpoint=handle_sse, methods=["GET"]),
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
# Define handler functions
async def handle_sse(request):
async with sse.connect_sse(
request.scope, request.receive, request._send
) as streams:
await app.run(
streams[0], streams[1], app.create_initialization_options()
)
# Return empty response to avoid NoneType error
return Response()
# Create an SSE transport at an endpoint
sse = mcp.server.sse.SseServerTransport("/messages/")
# Create Starlette routes for SSE and message handling
routes = [
Route("/sse", endpoint=handle_sse, methods=["GET"]),
Mount("/messages/", app=sse.handle_post_message),
]
if __name__ == "__main__":
# Create and run Starlette app
starlette_app = Starlette(routes=routes)
uvicorn.run(starlette_app, host="127.0.0.1")
run python
1
python3 mcp_server/my_adk_mcp_server.py
调试MCP服务
我们能通过postman调试mcp。只需要创建新请求的时候选择mcp
请求我们的mcp 服务之后，就能拉到所有的tools。点击tool就能执行tool。
结果是糟糕的，页面元素之复杂，导致返回的文字基本不可读。但是流程是通顺的。接下来的事情就是不停的优化这个tool的执行效果。
在agent里使用&调试
稍加修改我们在集成一个现成的MCP服务的程序，我们就能智能的使用这个tool了。
1
2
3
4
5
6
7
8
9
10
public static BaseAgent initAgent() {
return LlmAgent.builder()
.name(NAME)
.model("gemini-2.0-flash")
.description("Agent to get page content.")
.instruction(
"Use the 'load_web_page' tool to fetch content from a URL provided by the user.")
.tools(integrateJiraMCP())
.build();
}
这次我们用adk web 来进行调试，adk web会把所有的event记录清晰，有助于我们理解agent运行的机制。 java 启动adk-web 需要依赖sprint-boot，并且配置对应的agent 启动程序：
1
2
3
4
mvn exec:java \
-Dexec.mainClass="com.google.adk.web.AdkWebServer" \
-Dexec.args="--adk.agents.source-dir=src/main/java" \
-Dexec.classpathScope="compile"
执行，进入调试页面 ，输入prompt：
help me to get the web https://www.jakobhe.com/posts/after-reading-breath/
从左边的event栏，你很容易就能看到调用方法的情况。
一些想法
我们常常使用的是所谓的通用大模型，它适合用于处理一些通用和共识的问题，比如mysql，java这类技术问题，全世界的java 都有一样的knowledge base。
但是当我们要在垂直行业里，使用通用大模型时，企业就显得捉襟见肘了。
MCP恰恰就是提供这类垂直行业的执行能力的服务。比如绝大部分的软件从业人员，都有一套比较规范的流程，先从jira开始，到产品，到开发，到测试。
但是在餐饮行业，这个流程很可能是一个完完全全不一样的流程。 假设是：竞品调研，内部讨论，产品设计，开发，测试，发布。
恰好某餐饮公司，开发了一个竞品调研系统，那么竞品调研就可以是一个tools深度的集成在许多agent里，比如menu R&D流程的agent里，市场策略定制的agent里，等等。
总结一下子
今天我们一起做了：
- 实现了一个基本的MCP服务
- 利用postman调试MCP服务
- 集成MCP服务到agent里
- 利用adk web调试agent
- 从evnet来理解agent的运行机制