文章目錄
- MCP基礎概念
- 文件操作服務器
- 文件操作MCP接入谷歌ADK
- 項目創建
- 多輪對話代碼
MCP基礎概念
- MCP技術體系中，會將外部工具運行腳本稱作服務器，而接入這些外部工具的大模型運行環境稱作客戶端。
- 一個客戶端可以接入多個不同類型的服務器，但都要遵循
MCP通信協議。 - MCP服務器的輸出內容是一種標準格式的內容，只能被
MCP客戶端所識別。在客戶端和服務器都遵循MCP協議的時候，客戶端就能夠像Function calling中大模型調用外部工具一樣，調用MCP服務器里面的工具。
- 將
ADK作為客戶端，接入MCP工具。核心MCPToolset類，該類是ADK連接到MCP服務器的橋梁。 - 在實際過程中，ADK代理將執行以下操作
MCPToolset：- 連接：與
MCP服務器進程建立連接。該服務器可以是通過標準輸入/輸出 (StdioServerParameters) 進行通信的本地服務器，也可以是使用服務器發送事件 (SseServerParams) 的遠程服務器。 - 發現：查詢
MCP服務器以獲取可用工具（list_tools MCP方法）。 - 適應：將
MCP工具模式轉換為ADK兼容BaseTool實例。 - 公開：將這些適配的工具呈現給
ADK Agent。 - 代理調用：當
Agent決定使用其中一個工具時，發到MCP服務器并返回結果。MCPToolset將調用（call_tool MCP方法）轉 - 管理連接：處理與
MCP服務器進程的連接的生命周期，通常需要明確清理的周期（如進程結束后清理）。
- 連接：與
文件操作服務器
- Filesystem服務器是一個最基礎同時也是最常用的MCP服務器，同時也是官方推薦的服務器，服務器項目地址。
文件操作MCP接入谷歌ADK
項目創建
- 項目框架搭建：
# 初始化項目 uv init adk_mcp # 進入項目目錄 cd adk_mcp # 創建虛擬環境 uv venv # 激活虛擬環境 .venv\Scripts\activate # 安裝依賴 uv add google-adk litellm - 創建環境配置文件
.env：OPENAI_API_KEY=xxx OPENAI_API_BASE=https://api.openai-hk.com/v1 MODEL=openai/gpt-4o-mini - 創建
main.py，代碼內容如下：# 導入基礎庫 import os import asyncio from dotenv import load_dotenv from google.genai import types from google.adk.agents import Agent from google.adk.models.lite_llm import LiteLlm from google.adk.agents.llm_agent import LlmAgent from google.adk.runners import Runner from google.adk.sessions import InMemorySessionService from google.adk.tools.mcp_tool.mcp_toolset import MCPToolset, StdioServerParameters,StdioConnectionParamsDS_API_KEY = os.getenv("OPENAI_API_KEY") DS_BASE_URL = os.getenv("OPENAI_API_BASE") DS_MODEL = os.getenv("MODEL")model = LiteLlm(model=DS_MODEL,api_base=DS_BASE_URL,api_key=DS_API_KEY )# --- Step 1: 創建導入MCP工具函數 --- async def get_tools_async():"""Gets tools from the File System MCP Server."""print("Attempting to connect to MCP Filesystem server...")# 確保目錄存在test_dir = "D:\\Code\\adk_mcp\\test"if not os.path.exists(test_dir):os.makedirs(test_dir)print(f"Created directory: {test_dir}")try:# 創建MCP工具集實例mcp_toolset = MCPToolset(# 這里采用Studio通信方式進行調用connection_params=StdioConnectionParams(server_params=StdioServerParameters(command='npx',args=["-y", "@modelcontextprotocol/server-filesystem", test_dir],)))# 等待服務器啟動await asyncio.sleep(2)# 獲取工具列表tools = await mcp_toolset.get_tools()print("MCP Toolset created successfully.")print(f"Fetched {len(tools)} tools from MCP server.")# 返回工具和工具集對象用于清理return tools, mcp_toolsetexcept Exception as e:print(f"Error creating MCP tools: {e}")raise# --- Step 2: 創建ADK Agent -- async def get_agent_async():"""Creates an ADK Agent equipped with tools from the MCP Server."""try:tools, mcp_toolset = await get_tools_async()root_agent = Agent(model=model,name='filesystem_assistant',instruction='Help user interact with the local filesystem using available tools.',tools=tools, # 將MCP工具加載到Agent中)return root_agent, mcp_toolsetexcept Exception as e:print(f"Error creating agent: {e}")raise# --- Step 3: 執行主邏輯 --- async def async_main():mcp_toolset = Nonetry:# 創建會話管理器session_service = InMemorySessionService()session = await session_service.create_session(state={}, app_name='mcp_filesystem_app', user_id='user_fs')# 用戶輸入query = "請幫我查找目前文件夾里都有哪些文件？"print(f"User Query: '{query}'")content = types.Content(role='user', parts=[types.Part(text=query)])root_agent, mcp_toolset = await get_agent_async()runner = Runner(app_name='mcp_filesystem_app',agent=root_agent,session_service=session_service,)print("Running agent...")events_async = runner.run_async(session_id=session.id, user_id=session.user_id, new_message=content)async for event in events_async:print(f"Event received: {event}")except Exception as e:print(f"Error during main execution: {e}")raisefinally:# 運行完成后，關閉MCP服務器連接if mcp_toolset:print("Closing MCP server connection...")await mcp_toolset.close()print("Cleanup complete.")if __name__ == '__main__':try:asyncio.run(async_main())except Exception as e:print(f"An error occurred: {e}")
- 執行結果如下：
(adk_mcp) D:\Code\adk_mcp>uv run main.py D:\Code\adk_mcp\.venv\Lib\site-packages\pydantic\_internal\_fields.py:198: UserWarning: Field name "config_type" in "SequentialAgent" shadows an attribute in parent "BaseAgent"warnings.warn( User Query: '請幫我查找目前文件夾里都有哪些文件？' Attempting to connect to MCP Filesystem server... D:\Code\adk_mcp\.venv\Lib\site-packages\google\adk\tools\mcp_tool\mcp_tool.py:87: UserWarning: [EXPERIMENTAL] BaseAuthenticatedTool: This feature is experimental and may change or be removed in future versions without notice. It may introduce breaking changes at any time.super().__init__( auth_config or auth_config.auth_scheme is missing. Will skip authentication.Using FunctionTool instead if authentication is not required. MCP Toolset created successfully. Fetched 14 tools from MCP server. Running agent... Event received: content=Content(parts=[Part(text="""首先，我需要確定您當前的工作目錄位置。由于您沒有指定具體路徑，我將先獲取允許訪問的目錄列表："""),Part(function_call=FunctionCall(args={},id='call_5201be00eeb24a989be18f',name='list_allowed_directories')),],role='model' ) grounding_metadata=None partial=False turn_complete=None error_code=None error_message=None interrupted=None custom_metadata=None usage_metadata=GenerateContentResponseUsageMetadata(candidates_token_count=627,prompt_token_count=1806,total_token_count=2433 ) live_session_resumption_update=None invocation_id='e-1710d865-f562-4895-ae06-4a3f0155c138' author='filesystem_assistant' actions=EventActions(skip_summarization=None, state_delta={}, artifact_delta={}, transfer_to_agent=None, escalate=None, requested_auth_configs={}) long_running_tool_ids=set() branch=None id='71634100-b10a-4a81-b779-f34f52df12eb' timestamp=1754979735.932518 Event received: content=Content(parts=[Part(function_response=FunctionResponse(id='call_5201be00eeb24a989be18f',name='list_allowed_directories',response={'result': CallToolResult(content=[<... Max depth ...>,],isError=False)})),],role='user' ) grounding_metadata=None partial=None turn_complete=None error_code=None error_message=None interrupted=None custom_metadata=None usage_metadata=None live_session_resumption_update=None invocation_id='e-1710d865-f562-4895-ae06-4a3f0155c138' author='filesystem_assistant' actions=EventActions(skip_summarization=None, state_delta={}, artifact_delta={}, transfer_to_agent=None, escalate=None, requested_auth_configs={}) long_running_tool_ids=None branch=None id='00b7f823-5a66-493d-95bd-55ee93466fc2' timestamp=1754979761.37467 Event received: content=Content(parts=[Part(text="""根據系統配置，允許訪問的目錄是 `D:\Code\adk_mcp\test`。我將列出該目錄下的文件："""),Part(function_call=FunctionCall(args={'path': 'D:\\Code\\adk_mcp\\test'},id='call_2f66909ac2054847b6f9d3',name='list_directory')),],role='model' ) grounding_metadata=None partial=False turn_complete=None error_code=None error_message=None interrupted=None custom_metadata=None usage_metadata=GenerateContentResponseUsageMetadata(candidates_token_count=1105,prompt_token_count=1906,total_token_count=3011 ) live_session_resumption_update=None invocation_id='e-1710d865-f562-4895-ae06-4a3f0155c138' author='filesystem_assistant' actions=EventActions(skip_summarization=None, state_delta={}, artifact_delta={}, transfer_to_agent=None, escalate=None, requested_auth_configs={}) long_running_tool_ids=set() branch=None id='39358200-064d-478e-90f6-ab0855f13ec4' timestamp=1754979761.380695 Event received: content=Content(parts=[Part(function_response=FunctionResponse(id='call_2f66909ac2054847b6f9d3',name='list_directory',response={'result': CallToolResult(content=[<... Max depth ...>,],isError=False)})),],role='user' ) grounding_metadata=None partial=None turn_complete=None error_code=None error_message=None interrupted=None custom_metadata=None usage_metadata=None live_session_resumption_update=None invocation_id='e-1710d865-f562-4895-ae06-4a3f0155c138' author='filesystem_assistant' actions=EventActions(skip_summarization=None, state_delta={}, artifact_delta={}, transfer_to_agent=None, escalate=None, requested_auth_configs={}) long_running_tool_ids=None branch=None id='2d1197f7-c81d-41a5-9fc5-15652c50fa7e' timestamp=1754979803.754635 Event received: content=Content(parts=[Part(text="""當前目錄 `D:\Code\adk_mcp\test` 中的文件如下：- 📄 **系統架構設計師教程_帶目錄高清版.pdf**"""),],role='model' ) grounding_metadata=None partial=False turn_complete=None error_code=None error_message=None interrupted=None custom_metadata=None usage_metadata=GenerateContentResponseUsageMetadata(candidates_token_count=128,prompt_token_count=2021,total_token_count=2149 ) live_session_resumption_update=None invocation_id='e-1710d865-f562-4895-ae06-4a3f0155c138' author='filesystem_assistant' actions=EventActions(skip_summarization=None, state_delta={}, artifact_delta={}, transfer_to_agent=None, escalate=None, requested_auth_configs={}) long_running_tool_ids=None branch=None id='15fdd91c-e645-46fe-a613-76e1f28323c5' timestamp=1754979803.761915 Closing MCP server connection... Cleanup complete.
多輪對話代碼
- 以下代碼是
main.py的多輪對話的版本：# 導入基礎庫 import os import asyncio from dotenv import load_dotenv from google.genai import types from google.adk.agents import Agent from google.adk.models.lite_llm import LiteLlm from google.adk.agents.llm_agent import LlmAgent from google.adk.runners import Runner from google.adk.sessions import InMemorySessionService from google.adk.tools.mcp_tool.mcp_toolset import MCPToolset, StdioServerParameters,StdioConnectionParamsDS_API_KEY = os.getenv("OPENAI_API_KEY") DS_BASE_URL = os.getenv("OPENAI_API_BASE") DS_MODEL = os.getenv("MODEL")model = LiteLlm(model=DS_MODEL,api_base=DS_BASE_URL,api_key=DS_API_KEY )# --- Step 1: 創建導入MCP工具函數 --- async def get_tools_async():"""Gets tools from the File System MCP Server."""print("Attempting to connect to MCP Filesystem server...")# 確保目錄存在test_dir = "D:\\Code\\adk_mcp\\test"if not os.path.exists(test_dir):os.makedirs(test_dir)print(f"Created directory: {test_dir}")try:# 創建MCP工具集實例mcp_toolset = MCPToolset(# 這里采用Studio通信方式進行調用connection_params=StdioConnectionParams(server_params=StdioServerParameters(command='npx', # Command to run the serverargs=["-y", # Arguments for the command"@modelcontextprotocol/server-filesystem",test_dir],)))# 等待服務器啟動await asyncio.sleep(2)# 獲取工具列表tools = await mcp_toolset.get_tools()print("MCP Toolset created successfully.")print(f"Fetched {len(tools)} tools from MCP server.")# 返回工具和工具集對象用于清理return tools, mcp_toolsetexcept Exception as e:print(f"Error creating MCP tools: {e}")raise# --- Step 2: 創建ADK Agent -- async def get_agent_async():"""Creates an ADK Agent equipped with tools from the MCP Server."""try:tools, mcp_toolset = await get_tools_async()root_agent = Agent(model=model,name='filesystem_assistant',instruction='Help user interact with the local filesystem using available tools.',tools=tools, # 將MCP工具加載到Agent中)return root_agent, mcp_toolsetexcept Exception as e:print(f"Error creating agent: {e}")raise# 多輪對話函數 async def chat_loop(runner, user_id, session_id) -> None:print("\n🤖ADK + MCP對話已啟動！輸入'quit'退出。")while True:query = input("\n你: ").strip()if query.lower() == "quit":breaktry:print(f"\n>>> User Query: {query}")# Prepare the user's message in ADK formatcontent = types.Content(role='user', parts=[types.Part(text=query)])final_response_text = "Agent did not produce a final response." # Default# Key Concept: run_async executes the agent logic and yields Events.# We iterate through events to find the final answer.async for event in runner.run_async(user_id=user_id,session_id=session_id, new_message=content):# You can uncomment the line below to see *all* events during execution# print(f" [Event] Author: {event.author}, Type:{type(event).__name__}, Final: {event.is_final_response()}, Content:{event.content}")# Key Concept: is_final_response() marks the concluding message for the turn.if event.is_final_response():if event.content and event.content.parts:# Assuming text response in the first partfinal_response_text = event.content.parts[0].textelif event.actions and event.actions.escalate: # Handle potential errors / escalationsfinal_response_text = f"Agent escalated:{event.error_message or 'No specific message.'}"# Add more checks here if needed (e.g., specific error codes)breakprint(f"<<< Agent Response: {final_response_text}")except Exception as e:print(f"\n?調用過程出錯: {e}")# --- Step 3: 執行主邏輯 --- async def async_main():mcp_toolset = Nonetry:# 創建會話管理器session_service = InMemorySessionService()session = await session_service.create_session(state={}, app_name='mcp_filesystem_app', user_id='user_fs')# 用戶輸入query = "請幫我查找目前文件夾里都有哪些文件？"print(f"User Query: '{query}'")content = types.Content(role='user', parts=[types.Part(text=query)])root_agent, mcp_toolset = await get_agent_async()runner = Runner(app_name='mcp_filesystem_app',agent=root_agent,session_service=session_service,)print("Running agent...")events_async = runner.run_async(session_id=session.id, user_id=session.user_id, new_message=content)async for event in events_async:print(f"Event received: {event}")# 啟動多輪對話await chat_loop(runner, session.user_id, session.id)except Exception as e:print(f"Error during main execution: {e}")raisefinally:# 運行完成后，關閉MCP服務器連接if mcp_toolset:print("Closing MCP server connection...")await mcp_toolset.close()print("Cleanup complete.")if __name__ == '__main__':try:asyncio.run(async_main())except Exception as e:print(f"An error occurred: {e}")
