diff --git a/server/routers/chat_router.py b/server/routers/chat_router.py index b3201d34..45f58e32 100644 --- a/server/routers/chat_router.py +++ b/server/routers/chat_router.py @@ -185,8 +185,8 @@ async def get_tools(agent_id: str, current_user: User = Depends(get_required_use if not (agent := agent_manager.get_agent(agent_id)): raise HTTPException(status_code=404, detail=f"智能体 {agent_id} 不存在") - if hasattr(agent, "agent_tools"): - tools = agent.agent_tools + if hasattr(agent, "get_tools"): + tools = agent.get_tools() else: tools = get_buildin_tools() diff --git a/server/utils/common_utils.py b/server/utils/common_utils.py index 94791e00..73cce32c 100644 --- a/server/utils/common_utils.py +++ b/server/utils/common_utils.py @@ -52,3 +52,14 @@ def log_operation(db: Session, user_id: int, operation: str, details: str = None def get_user_dict(user: User, include_password: bool = False) -> dict: """获取用户字典表示""" return user.to_dict(include_password) + + +def convert_serializable(obj): + """将对象转换为可序列化的格式""" + if isinstance(obj, list | tuple): + return [convert_serializable(item) for item in obj] + if isinstance(obj, dict): + return {k: convert_serializable(v) for k, v in obj.items()} + if hasattr(obj, '__dict__'): + return convert_serializable(vars(obj)) + return obj diff --git a/src/agents/chatbot/context.py b/src/agents/chatbot/context.py index 797565d1..83abeb44 100644 --- a/src/agents/chatbot/context.py +++ b/src/agents/chatbot/context.py @@ -3,6 +3,7 @@ from dataclasses import dataclass, field from src.agents.common.context import BaseContext from src.agents.common.tools import gen_tool_info +from src.agents.common.mcp import MCP_SERVERS from .tools import get_tools @@ -26,3 +27,12 @@ class Context(BaseContext): "description": "工具列表" }, ) + + mcps: list[str] = field( + default_factory=list, + metadata={ + "name": "MCP服务器", + "options": list(MCP_SERVERS.keys()), + "description": "MCP服务器列表" + }, + ) diff --git a/src/agents/chatbot/graph.py b/src/agents/chatbot/graph.py index ca89eeee..5ea07342 100644 --- a/src/agents/chatbot/graph.py +++ b/src/agents/chatbot/graph.py @@ -17,6 +17,8 @@ from src.utils import logger from src.agents.common.utils import get_cur_time_with_utc from src.agents.common.base import BaseAgent from src.agents.common.models import load_chat_model +from src.agents.common.mcp import get_mcp_tools + from .state import State from .context import Context @@ -34,38 +36,44 @@ class ChatbotAgent(BaseAgent): self.context_schema = Context self.workdir = Path(sys_config.save_dir) / "agents" / self.module_name self.workdir.mkdir(parents=True, exist_ok=True) - self.agent_tools = get_tools() + self.agent_tools = None - def _get_tools(self, tools: list[str]): + def get_tools(self): + return get_tools() + + async def _get_invoke_tools(self, selected_tools: list[str], selected_mcps: list[str]): """根据配置获取工具。 默认不使用任何工具。 如果配置为列表,则使用列表中的工具。 """ - self.agent_tools = get_tools() - if tools is None or not isinstance(tools, list) or len(tools) == 0: - # 默认不使用任何工具 - logger.info("未配置工具或配置为空,不使用任何工具") - return [] - else: + enabled_tools = [] + self.agent_tools = self.agent_tools or self.get_tools() + if selected_tools and isinstance(selected_tools, list) and len(selected_tools) > 0: # 使用配置中指定的工具 - tools = [tool for tool in self.agent_tools if tool.name in tools] - logger.info(f"使用工具: {[tool.name for tool in tools]}") - return tools + enabled_tools = [tool for tool in self.agent_tools if tool.name in selected_tools] + + if selected_mcps and isinstance(selected_mcps, list) and len(selected_mcps) > 0: + for mcp in selected_mcps: + enabled_tools.extend(await get_mcp_tools(mcp)) + + return enabled_tools async def llm_call(self, state: State, runtime: Runtime[Context] = None) -> dict[str, Any]: """调用 llm 模型 - 异步版本以支持异步工具""" - system_prompt = f"{runtime.context.system_prompt}. Current time is {get_cur_time_with_utc()}" model = load_chat_model(runtime.context.model) # 这里要根据配置动态获取工具 - if tools := self._get_tools(runtime.context.tools): - model = model.bind_tools(tools) + available_tools = await self._get_invoke_tools(runtime.context.tools, runtime.context.mcps) + logger.info(f"LLM binded ({len(available_tools)}) available_tools: {[tool.name for tool in available_tools]}") + + if available_tools: + model = model.bind_tools(available_tools) # 使用异步调用 response = cast( AIMessage, await model.ainvoke( - [{"role": "system", "content": system_prompt}, *state.messages] + [{"role": "system", "content": runtime.context.system_prompt}, *state.messages] ), ) return {"messages": [response]} @@ -80,7 +88,7 @@ class ChatbotAgent(BaseAgent): and executes the requested tool calls from the last message. """ # Get available tools based on configuration - available_tools = get_tools() + available_tools = await self._get_invoke_tools(runtime.context.tools, runtime.context.mcps) # Create a ToolNode with the available tools tool_node = ToolNode(available_tools) @@ -92,11 +100,8 @@ class ChatbotAgent(BaseAgent): async def get_graph(self, **kwargs): """构建图""" - if self.graph: - return self.graph - - runnable_tools = get_tools() - logger.debug(f"build graph `{self.id}` with {len(runnable_tools)} tools") + # if self.graph: + # return self.graph builder = StateGraph(State, context_schema=self.context_schema) builder.add_node("chatbot", self.llm_call) diff --git a/src/agents/chatbot/tools.py b/src/agents/chatbot/tools.py index 49d59c04..1d8fb52a 100644 --- a/src/agents/chatbot/tools.py +++ b/src/agents/chatbot/tools.py @@ -30,5 +30,5 @@ def get_tools() -> dict[str, Any]: """获取所有可运行的工具(给大模型使用)""" tools = get_buildin_tools() tools.append(calculator) - return tools + diff --git a/src/agents/common/base.py b/src/agents/common/base.py index c4a079a6..41a551f4 100644 --- a/src/agents/common/base.py +++ b/src/agents/common/base.py @@ -36,7 +36,6 @@ class BaseAgent: "name": self.name if hasattr(self, "name") else "Unknown", "description": self.description if hasattr(self, "description") else "Unknown", "configurable_items": self.context_schema.get_configurable_items(), - "all_tools": self.all_tools if hasattr(self, "all_tools") else [], "has_checkpointer": await self.check_checkpointer(), } @@ -51,9 +50,8 @@ class BaseAgent: async def stream_messages(self, messages: list[str], input_context = None, **kwargs): graph = await self.get_graph() - logger.debug(f"stream_messages: {input_context}") - context = self.context_schema.from_file(module_name=self.module_name, input_context=input_context) + logger.debug(f"stream_messages: {context}") # TODO 的 Checkpointer 似乎还没有适配最新的 Context API async for msg, metadata in graph.astream({"messages": messages}, stream_mode="messages", context=context, config={"configurable": input_context}): yield msg, metadata diff --git a/src/agents/common/mcp.py b/src/agents/common/mcp.py new file mode 100644 index 00000000..e1e00829 --- /dev/null +++ b/src/agents/common/mcp.py @@ -0,0 +1,100 @@ +"""MCP Client setup and management for LangGraph ReAct Agent.""" + +import traceback +from typing import Any, cast +from collections.abc import Callable + +from langchain_mcp_adapters.tools import load_mcp_tools +from langchain_mcp_adapters.client import ( # type: ignore[import-untyped] + MultiServerMCPClient, +) + +from src.utils import logger + +# Global MCP client and tools cache +_mcp_client: MultiServerMCPClient | None = None +_mcp_tools_cache: dict[str, list[Callable[..., Any]]] = {} + +# MCP Server configurations +MCP_SERVERS = { + "deepwiki": { + "url": "https://mcp.deepwiki.com/mcp", + "transport": "streamable_http", + }, + # Add more MCP servers here as needed + "sequentialthinking": { + "url": "https://remote.mcpservers.org/sequentialthinking/mcp", + "transport": "streamable_http", + }, + "time": { + "command": "uvx", + "args": ["mcp-server-time"], + "transport": "stdio", + } +} + + +async def get_mcp_client( + server_configs: dict[str, Any] | None = None, +) -> MultiServerMCPClient | None: + """Get or initialize the global MCP client with given server configurations.""" + global _mcp_client + + if _mcp_client is None: + configs = server_configs or MCP_SERVERS + try: + _mcp_client = MultiServerMCPClient(configs) # pyright: ignore[reportArgumentType] + logger.info(f"Initialized MCP client with servers: {list(configs.keys())}") + except Exception as e: + logger.error("Failed to initialize MCP client: %s", e) + return None + return _mcp_client + + +async def get_mcp_tools(server_name: str) -> list[Callable[..., Any]]: + """Get MCP tools for a specific server, initializing client if needed.""" + global _mcp_tools_cache + + # Return cached tools if available + if server_name in _mcp_tools_cache: + return _mcp_tools_cache[server_name] + + try: + client = await get_mcp_client({server_name: MCP_SERVERS[server_name]}) + if client is None: + _mcp_tools_cache[server_name] = [] + return [] + + # Get all tools and filter by server (if tools have server metadata) + all_tools = await client.get_tools() + tools = cast(list[Callable[..., Any]], all_tools) + + _mcp_tools_cache[server_name] = tools + logger.info(f"Loaded {len(tools)} tools from MCP server '{server_name}'") + return tools + except Exception as e: + logger.warning(f"Failed to load tools from MCP server '{server_name}': %s\n{traceback.format_exc()}", e) + _mcp_tools_cache[server_name] = [] + return [] + +async def get_all_mcp_tools() -> list[Callable[..., Any]]: + """Get all tools from all configured MCP servers.""" + all_tools = [] + for server_name in MCP_SERVERS.keys(): + tools = await get_mcp_tools(server_name) + all_tools.extend(tools) + return all_tools + + +def add_mcp_server(name: str, config: dict[str, Any]) -> None: + """Add a new MCP server configuration.""" + MCP_SERVERS[name] = config + # Clear client to force reinitialization with new config + clear_mcp_cache() + + +def clear_mcp_cache() -> None: + """Clear the MCP client and tools cache (useful for testing).""" + global _mcp_client, _mcp_tools_cache + _mcp_client = None + _mcp_tools_cache = {} diff --git a/src/agents/common/tools.py b/src/agents/common/tools.py index 69532ada..ef902058 100644 --- a/src/agents/common/tools.py +++ b/src/agents/common/tools.py @@ -1,7 +1,7 @@ +import inspect import asyncio import traceback from typing import Annotated, Any - from pydantic import BaseModel, Field from langchain_core.tools import StructuredTool, tool from langchain_tavily import TavilySearch @@ -22,7 +22,7 @@ def query_knowledge_graph(query: Annotated[str, "The keyword to query knowledge logger.error(f"Knowledge graph query error: {e}, {traceback.format_exc()}") return f"知识图谱查询失败: {str(e)}" -def get_static_tools() -> dict[str, Any]: +def get_static_tools() -> list: """注册静态工具""" static_tools = [ query_knowledge_graph, @@ -45,8 +45,7 @@ class KnowledgeRetrieverModel(BaseModel): - -def get_kb_based_tools() -> dict[str, Any]: +def get_kb_based_tools() -> list: """获取所有知识库基于的工具""" # 获取所有知识库 kb_tools = [] @@ -80,7 +79,7 @@ def get_kb_based_tools() -> dict[str, Any]: # 构建工具描述 description = ( f"使用 {retrieve_info['name']} 知识库进行检索。\n" - f"下面是这个知识库的描述:\n{retrieve_info['description'] or '没有描述。'}" + f"下面是这个知识库的描述:\n{retrieve_info['description'] or '没有描述。'} " ) # 使用工厂函数创建检索器包装函数,避免闭包问题 @@ -107,7 +106,7 @@ def get_kb_based_tools() -> dict[str, Any]: return kb_tools -def get_buildin_tools() -> dict[str, Any]: +def get_buildin_tools() -> list: """获取所有可运行的工具(给大模型使用)""" tools = [] @@ -119,11 +118,10 @@ def get_buildin_tools() -> dict[str, Any]: except Exception as e: logger.error(f"Failed to get knowledge base retrievers: {e}") - logger.info(f"Total tools available: {len(tools)}") return tools -def gen_tool_info(tools) -> dict[str, dict[str, Any]]: +def gen_tool_info(tools) -> list[dict[str, Any]]: """获取所有工具的信息(用于前端展示)""" tools_info = [] @@ -137,29 +135,29 @@ def gen_tool_info(tools) -> dict[str, dict[str, Any]]: "name": metadata.get('name', tool_obj.name), "description": tool_obj.description, 'metadata': metadata, - "args": [] + "args": [], + # "is_async": is_async # Include async information } - schema = tool_obj.args_schema.schema() - for arg_name, arg_info in schema['properties'].items(): - info["args"].append({ - "name": arg_name, - "type": arg_info.get('type', ''), - "description": arg_info.get('description', '') - }) + if hasattr(tool_obj, 'args_schema') and tool_obj.args_schema: + schema = tool_obj.args_schema.schema() + for arg_name, arg_info in schema.get('properties', {}).items(): + info["args"].append({ + "name": arg_name, + "type": arg_info.get('type', ''), + "description": arg_info.get('description', '') + }) tools_info.append(info) # logger.debug(f"Successfully processed tool info for {tool_obj.name}") except Exception as e: - logger.error(f"Failed to process tool {tool_obj.name}: {e}") + logger.error(f"Failed to process tool {getattr(tool_obj, 'name', 'unknown')}: {e}\n{traceback.format_exc()}") continue except Exception as e: - logger.error(f"Failed to get tools info: {e}") - return {} + logger.error(f"Failed to get tools info: {e}\n{traceback.format_exc()}") + return [] logger.info(f"Successfully extracted info for {len(tools_info)} tools") return tools_info - - diff --git a/web/src/components/AgentMessageComponent.vue b/web/src/components/AgentMessageComponent.vue index 7c9162ca..0bc4d511 100644 --- a/web/src/components/AgentMessageComponent.vue +++ b/web/src/components/AgentMessageComponent.vue @@ -377,19 +377,10 @@ const toggleToolCall = (toolCallId) => { transition: all 0.3s ease; .tool-params { - padding: 10px 16px; + padding: 8px 12px; background-color: var(--gray-25); border-bottom: 1px solid var(--gray-150); - .tool-params-header { - // unused - background-color: var(--gray-100); - font-size: 13px; - color: var(--gray-800); - margin-bottom: 8px; - font-weight: 500; - } - .tool-params-content { margin: 0; font-size: 13px; diff --git a/web/src/components/ToolCallingResult/ToolResultRenderer.vue b/web/src/components/ToolCallingResult/ToolResultRenderer.vue index 9e05466f..a478547a 100644 --- a/web/src/components/ToolCallingResult/ToolResultRenderer.vue +++ b/web/src/components/ToolCallingResult/ToolResultRenderer.vue @@ -27,9 +27,9 @@
-
+
{{ formatData(parsedData) }}