diff --git a/pyproject.toml b/pyproject.toml index 953dec12..6deba34a 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -23,7 +23,7 @@ dependencies = [ "pynacl>=1.5.0", # Required for Discord voice support "gewechat-client>=0.1.5", "lark-oapi>=1.4.15", - "mcp>=1.8.1", + "mcp>=1.20.0", "nakuru-project-idk>=0.0.2.1", "ollama>=0.4.8", "openai>1.0.0", diff --git a/src/langbot/pkg/entity/persistence/mcp.py b/src/langbot/pkg/entity/persistence/mcp.py index 74478dc7..e9eedbdb 100644 --- a/src/langbot/pkg/entity/persistence/mcp.py +++ b/src/langbot/pkg/entity/persistence/mcp.py @@ -9,7 +9,7 @@ class MCPServer(Base): uuid = sqlalchemy.Column(sqlalchemy.String(255), primary_key=True, unique=True) name = sqlalchemy.Column(sqlalchemy.String(255), nullable=False) enable = sqlalchemy.Column(sqlalchemy.Boolean, nullable=False, default=False) - mode = sqlalchemy.Column(sqlalchemy.String(255), nullable=False) # stdio, sse + mode = sqlalchemy.Column(sqlalchemy.String(255), nullable=False) # stdio, sse, http extra_args = sqlalchemy.Column(sqlalchemy.JSON, nullable=False, default={}) created_at = sqlalchemy.Column(sqlalchemy.DateTime, nullable=False, server_default=sqlalchemy.func.now()) updated_at = sqlalchemy.Column( diff --git a/src/langbot/pkg/provider/runners/localagent.py b/src/langbot/pkg/provider/runners/localagent.py index 4197f076..b335ed11 100644 --- a/src/langbot/pkg/provider/runners/localagent.py +++ b/src/langbot/pkg/provider/runners/localagent.py @@ -215,16 +215,24 @@ class LocalAgentRunner(runner.RequestRunner): parameters = json.loads(func.arguments) func_ret = await self.ap.tool_mgr.execute_func_call(func.name, parameters, query=query) + + # Handle return value content + tool_content = None + if isinstance(func_ret, list) and len(func_ret) > 0 and isinstance(func_ret[0], provider_message.ContentElement): + tool_content = func_ret + else: + tool_content = json.dumps(func_ret, ensure_ascii=False) + if is_stream: msg = provider_message.MessageChunk( role='tool', - content=json.dumps(func_ret, ensure_ascii=False), + content=tool_content, tool_call_id=tool_call.id, ) else: msg = provider_message.Message( role='tool', - content=json.dumps(func_ret, ensure_ascii=False), + content=tool_content, tool_call_id=tool_call.id, ) diff --git a/src/langbot/pkg/provider/tools/loaders/mcp.py b/src/langbot/pkg/provider/tools/loaders/mcp.py index 79afcc4f..4b0583c6 100644 --- a/src/langbot/pkg/provider/tools/loaders/mcp.py +++ b/src/langbot/pkg/provider/tools/loaders/mcp.py @@ -7,14 +7,18 @@ import traceback from langbot_plugin.api.entities.events import pipeline_query import sqlalchemy import asyncio +import httpx + import uuid as uuid_module from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client from mcp.client.sse import sse_client +from mcp.client.streamable_http import streamable_http_client from .. import loader from ....core import app import langbot_plugin.api.entities.builtin.resource.tool as resource_tool +import langbot_plugin.api.entities.builtin.provider.message as provider_message from ....entity.persistence import mcp as persistence_mcp @@ -35,7 +39,7 @@ class RuntimeMCPSession: server_config: dict - session: ClientSession + session: ClientSession | None exit_stack: AsyncExitStack @@ -52,6 +56,8 @@ class RuntimeMCPSession: _ready_event: asyncio.Event + error_message: str | None = None + def __init__(self, server_name: str, server_config: dict, enable: bool, ap: app.Application): self.server_name = server_name self.server_uuid = server_config.get('uuid', '') @@ -100,6 +106,24 @@ class RuntimeMCPSession: await self.session.initialize() + async def _init_streamable_http_server(self): + transport = await self.exit_stack.enter_async_context( + streamable_http_client( + self.server_config['url'], + http_client=httpx.AsyncClient( + headers=self.server_config.get('headers', {}), + timeout=self.server_config.get('timeout', 10), + follow_redirects=True, + ), + ) + ) + + read, write, _ = transport + + self.session = await self.exit_stack.enter_async_context(ClientSession(read, write)) + + await self.session.initialize() + async def _lifecycle_loop(self): """在后台任务中管理整个MCP会话的生命周期""" try: @@ -107,6 +131,8 @@ class RuntimeMCPSession: await self._init_stdio_python_server() elif self.server_config['mode'] == 'sse': await self._init_sse_server() + elif self.server_config['mode'] == 'http': + await self._init_streamable_http_server() else: raise ValueError(f'无法识别 MCP 服务器类型: {self.server_name}: {self.server_config}') @@ -122,6 +148,7 @@ class RuntimeMCPSession: except Exception as e: self.status = MCPSessionStatus.ERROR + self.error_message = str(e) self.ap.logger.error(f'Error in MCP session lifecycle {self.server_name}: {e}\n{traceback.format_exc()}') # 即使出错也要设置ready事件，让start()方法知道初始化已完成 self._ready_event.set() @@ -154,6 +181,9 @@ class RuntimeMCPSession: raise Exception('Connection failed, please check URL') async def refresh(self): + if not self.session: + return + self.functions.clear() tools = await self.session.list_tools() @@ -163,18 +193,36 @@ class RuntimeMCPSession: for tool in tools.tools: async def func(*, _tool=tool, **kwargs): + if not self.session: + raise Exception("MCP session is not connected") + result = await self.session.call_tool(_tool.name, kwargs) if result.isError: - raise Exception(result.content[0].text) - return result.content[0].text + error_texts = [] + for content in result.content: + if content.type == 'text': + error_texts.append(content.text) + raise Exception("\n".join(error_texts) if error_texts else "Unknown error from MCP tool") + + result_contents: list[provider_message.ContentElement] = [] + for content in result.content: + if content.type == 'text': + result_contents.append(provider_message.ContentElement.from_text(content.text)) + elif content.type == 'image': + result_contents.append(provider_message.ContentElement.from_image_base64(content.image_base64)) + elif content.type == 'resource': + # TODO: Handle resource content + pass + + return result_contents func.__name__ = tool.name self.functions.append( resource_tool.LLMTool( name=tool.name, - human_desc=tool.description, - description=tool.description, + human_desc=tool.description or "", + description=tool.description or "", parameters=tool.inputSchema, func=func, ) @@ -186,6 +234,7 @@ class RuntimeMCPSession: def get_runtime_info_dict(self) -> dict: return { 'status': self.status.value, + 'error_message': self.error_message, 'tool_count': len(self.get_tools()), 'tools': [ { diff --git a/web/src/app/home/plugins/mcp-server/mcp-card/MCPCardComponent.tsx b/web/src/app/home/plugins/mcp-server/mcp-card/MCPCardComponent.tsx index 2ae43a9c..2647fe05 100644 --- a/web/src/app/home/plugins/mcp-server/mcp-card/MCPCardComponent.tsx +++ b/web/src/app/home/plugins/mcp-server/mcp-card/MCPCardComponent.tsx @@ -3,6 +3,7 @@ import { useState, useEffect } from 'react'; import { httpClient } from '@/app/infra/http/HttpClient'; import { Switch } from '@/components/ui/switch'; import { Button } from '@/components/ui/button'; +import { Badge } from '@/components/ui/badge'; import { toast } from 'sonner'; import { useTranslation } from 'react-i18next'; import { RefreshCcw, Wrench, Ban, AlertCircle, Loader2 } from 'lucide-react'; @@ -98,9 +99,14 @@ export default function MCPCardComponent({
-
-
- {cardVO.name} +
+
+
+ {cardVO.name} +
+ + {cardVO.mode.toUpperCase()} +
diff --git a/web/src/app/home/plugins/mcp-server/mcp-form/MCPFormDialog.tsx b/web/src/app/home/plugins/mcp-server/mcp-form/MCPFormDialog.tsx index 72901d87..f1d647a8 100644 --- a/web/src/app/home/plugins/mcp-server/mcp-form/MCPFormDialog.tsx +++ b/web/src/app/home/plugins/mcp-server/mcp-form/MCPFormDialog.tsx @@ -43,6 +43,9 @@ import { MCPTool, MCPServer, MCPSessionStatus, + MCPServerExtraArgsSSE, + MCPServerExtraArgsHttp, + MCPServerExtraArgsStdio, } from '@/app/infra/entities/api'; import { CustomApiError } from '@/app/infra/entities/common'; @@ -133,11 +136,11 @@ function StatusDisplay({ {t('mcp.connectionFailed')}
- {/* {runtimeInfo.error_message && ( + {runtimeInfo.error_message && (
{runtimeInfo.error_message}
- )} */} + )}
); } @@ -163,31 +166,52 @@ function ToolsList({ tools }: { tools: MCPTool[] }) { } const getFormSchema = (t: (key: string) => string) => - z.object({ - name: z - .string({ required_error: t('mcp.nameRequired') }) - .min(1, { message: t('mcp.nameRequired') }), - timeout: z - .number({ invalid_type_error: t('mcp.timeoutMustBeNumber') }) - .positive({ message: t('mcp.timeoutMustBePositive') }) - .default(30), - ssereadtimeout: z - .number({ invalid_type_error: t('mcp.sseTimeoutMustBeNumber') }) - .positive({ message: t('mcp.timeoutMustBePositive') }) - .default(300), - url: z - .string({ required_error: t('mcp.urlRequired') }) - .min(1, { message: t('mcp.urlRequired') }), - extra_args: z - .array( - z.object({ - key: z.string(), - type: z.enum(['string', 'number', 'boolean']), - value: z.string(), - }), - ) - .optional(), - }); + z + .object({ + name: z + .string({ required_error: t('mcp.nameRequired') }) + .min(1, { message: t('mcp.nameRequired') }), + mode: z.enum(['sse', 'stdio', 'http']), + timeout: z + .number({ invalid_type_error: t('mcp.timeoutMustBeNumber') }) + .positive({ message: t('mcp.timeoutMustBePositive') }) + .default(30), + ssereadtimeout: z + .number({ invalid_type_error: t('mcp.sseTimeoutMustBeNumber') }) + .positive({ message: t('mcp.timeoutMustBePositive') }) + .default(300), + url: z.string().optional(), + command: z.string().optional(), + args: z.array(z.object({ value: z.string() })).optional(), + extra_args: z + .array( + z.object({ + key: z.string(), + type: z.enum(['string', 'number', 'boolean']), + value: z.string(), + }), + ) + .optional(), + }) + .superRefine((data, ctx) => { + if (data.mode === 'sse' || data.mode === 'http') { + if (!data.url || data.url.length === 0) { + ctx.addIssue({ + code: z.ZodIssueCode.custom, + message: t('mcp.urlRequired'), + path: ['url'], + }); + } + } else if (data.mode === 'stdio') { + if (!data.command || data.command.length === 0) { + ctx.addIssue({ + code: z.ZodIssueCode.custom, + message: t('mcp.commandRequired'), + path: ['command'], + }); + } + } + }); type FormValues = z.infer> & { timeout: number; @@ -218,7 +242,10 @@ export default function MCPFormDialog({ resolver: zodResolver(formSchema) as unknown as Resolver, defaultValues: { name: '', + mode: 'sse', url: '', + command: '', + args: [], timeout: 30, ssereadtimeout: 300, extra_args: [], @@ -228,20 +255,33 @@ export default function MCPFormDialog({ const [extraArgs, setExtraArgs] = useState< { key: string; type: 'string' | 'number' | 'boolean'; value: string }[] >([]); + const [stdioArgs, setStdioArgs] = useState<{ value: string }[]>([]); const [mcpTesting, setMcpTesting] = useState(false); const [runtimeInfo, setRuntimeInfo] = useState( null, ); const pollingIntervalRef = useRef(null); + const watchMode = form.watch('mode'); + // Load server data when editing useEffect(() => { if (open && isEditMode && serverName) { loadServerForEdit(serverName); } else if (open && !isEditMode) { // Reset form when creating new server - form.reset(); + form.reset({ + name: '', + mode: 'sse', + url: '', + command: '', + args: [], + timeout: 30, + ssereadtimeout: 300, + extra_args: [], + }); setExtraArgs([]); + setStdioArgs([]); setRuntimeInfo(null); } @@ -291,25 +331,49 @@ export default function MCPFormDialog({ const resp = await httpClient.getMCPServer(serverName); const server = resp.server ?? resp; - const extraArgs = server.extra_args; form.setValue('name', server.name); - form.setValue('url', extraArgs.url); - form.setValue('timeout', extraArgs.timeout); - form.setValue('ssereadtimeout', extraArgs.ssereadtimeout); + form.setValue('mode', server.mode); - if (extraArgs.headers) { - const headers = Object.entries(extraArgs.headers).map( - ([key, value]) => ({ - key, - type: 'string' as const, - value: String(value), - }), - ); - setExtraArgs(headers); - form.setValue('extra_args', headers); + if (server.mode === 'sse' || server.mode === 'http') { + form.setValue('url', server.extra_args.url); + form.setValue('timeout', server.extra_args.timeout); + + if (server.mode === 'sse') { + form.setValue('ssereadtimeout', server.extra_args.ssereadtimeout); + } + + if (server.extra_args.headers) { + const headers = Object.entries(server.extra_args.headers).map( + ([key, value]) => ({ + key, + type: 'string' as const, + value: String(value), + }), + ); + setExtraArgs(headers); + form.setValue('extra_args', headers); + } + } else if (server.mode === 'stdio') { + form.setValue('command', server.extra_args.command); + const args = (server.extra_args.args || []).map((arg: string) => ({ + value: arg, + })); + setStdioArgs(args); + form.setValue('args', args); + + if (server.extra_args.env) { + const envs = Object.entries(server.extra_args.env).map( + ([key, value]) => ({ + key, + type: 'string' as const, + value: String(value), + }), + ); + setExtraArgs(envs); + form.setValue('extra_args', envs); + } } - // Set runtime_info from server data if (server.runtime_info) { setRuntimeInfo(server.runtime_info); } else { @@ -322,28 +386,60 @@ export default function MCPFormDialog({ } async function handleFormSubmit(value: z.infer) { - // Convert extra_args to headers - all values must be strings according to MCPServerExtraArgsSSE - const headers: Record = {}; - value.extra_args?.forEach((arg) => { - // Convert all values to strings to match MCPServerExtraArgsSSE.headers type - headers[arg.key] = String(arg.value); - }); - try { - const serverConfig: Omit< - MCPServer, - 'uuid' | 'created_at' | 'updated_at' | 'runtime_info' - > = { - name: value.name, - mode: 'sse' as const, - enable: true, - extra_args: { - url: value.url, - headers: headers, - timeout: value.timeout, - ssereadtimeout: value.ssereadtimeout, - }, - }; + let serverConfig: MCPServer; + + if (value.mode === 'sse' || value.mode === 'http') { + const headers: Record = {}; + value.extra_args?.forEach((arg) => { + headers[arg.key] = String(arg.value); + }); + + if (value.mode === 'sse') { + serverConfig = { + name: value.name, + mode: 'sse', + enable: true, + extra_args: { + url: value.url!, + headers: headers, + timeout: value.timeout, + ssereadtimeout: value.ssereadtimeout, + }, + }; + } else { + serverConfig = { + name: value.name, + mode: 'http', + enable: true, + extra_args: { + url: value.url!, + headers: headers, + timeout: value.timeout, + }, + }; + } + } else { + // Convert extra_args to env + const env: Record = {}; + value.extra_args?.forEach((arg) => { + env[arg.key] = String(arg.value); + }); + + // Convert args object array to string array + const args = value.args?.map((arg) => arg.value) || []; + + serverConfig = { + name: value.name, + mode: 'stdio', + enable: true, + extra_args: { + command: value.command!, + args: args, + env: env, + }, + }; + } if (isEditMode && serverName) { await httpClient.updateMCPServer(serverName, serverConfig); @@ -365,19 +461,44 @@ export default function MCPFormDialog({ setMcpTesting(true); try { - const { task_id } = await httpClient.testMCPServer('_', { - name: form.getValues('name'), - mode: 'sse', - enable: true, - extra_args: { - url: form.getValues('url'), + const mode = form.getValues('mode'); + let extraArgsData: + | MCPServerExtraArgsSSE + | MCPServerExtraArgsHttp + | MCPServerExtraArgsStdio; + + if (mode === 'sse') { + extraArgsData = { + url: form.getValues('url')!, timeout: form.getValues('timeout'), - ssereadtimeout: form.getValues('ssereadtimeout'), headers: Object.fromEntries( extraArgs.map((arg) => [arg.key, arg.value]), ), - }, - }); + ssereadtimeout: form.getValues('ssereadtimeout'), + }; + } else if (mode === 'http') { + extraArgsData = { + url: form.getValues('url')!, + timeout: form.getValues('timeout'), + headers: Object.fromEntries( + extraArgs.map((arg) => [arg.key, arg.value]), + ), + }; + } else { + extraArgsData = { + command: form.getValues('command')!, + args: stdioArgs.map((arg) => arg.value), + env: Object.fromEntries(extraArgs.map((arg) => [arg.key, arg.value])), + }; + } + + const { task_id } = await httpClient.testMCPServer('_', { + name: form.getValues('name'), + mode: mode, + enable: true, + extra_args: extraArgsData, + } as MCPServer); + if (!task_id) { throw new Error(t('mcp.noTaskId')); } @@ -448,11 +569,31 @@ export default function MCPFormDialog({ form.setValue('extra_args', newArgs); }; + const addStdioArg = () => { + const newArgs = [...stdioArgs, { value: '' }]; + setStdioArgs(newArgs); + form.setValue('args', newArgs); + }; + + const removeStdioArg = (index: number) => { + const newArgs = stdioArgs.filter((_, i) => i !== index); + setStdioArgs(newArgs); + form.setValue('args', newArgs); + }; + + const updateStdioArg = (index: number, value: string) => { + const newArgs = [...stdioArgs]; + newArgs[index] = { value }; + setStdioArgs(newArgs); + form.setValue('args', newArgs); + }; + const handleDialogClose = (open: boolean) => { onOpenChange(open); if (!open) { form.reset(); setExtraArgs([]); + setStdioArgs([]); setRuntimeInfo(null); } }; @@ -518,58 +659,155 @@ export default function MCPFormDialog({ ( - {t('mcp.url')} - - - + {t('mcp.serverMode')} + )} /> - ( - - {t('mcp.timeout')} - - field.onChange(Number(e.target.value))} - /> - - - - )} - /> + {(watchMode === 'sse' || watchMode === 'http') && ( + <> + ( + + {t('mcp.url')} + + + + + + )} + /> + + ( + + {t('mcp.timeout')} + + + field.onChange(Number(e.target.value)) + } + /> + + + + )} + /> + + {watchMode === 'sse' && ( + ( + + {t('mcp.sseTimeout')} + + + field.onChange(Number(e.target.value)) + } + /> + + + + )} + /> + )} + + )} + + {watchMode === 'stdio' && ( + <> + ( + + {t('mcp.command')} + + + + + + )} + /> - ( - {t('mcp.sseTimeout')} - - field.onChange(Number(e.target.value))} - /> - - + {t('mcp.args')} +
+ {stdioArgs.map((arg, index) => ( +
+ + updateStdioArg(index, e.target.value) + } + /> + +
+ ))} + +
- )} - /> + + )} - {t('models.extraParameters')} + + {watchMode === 'sse' || watchMode === 'http' + ? t('mcp.headers') + : t('mcp.env')} +
{extraArgs.map((arg, index) => (
@@ -580,7 +818,12 @@ export default function MCPFormDialog({ updateExtraArg(index, 'key', e.target.value) } /> - + */} ))}
diff --git a/web/src/app/infra/entities/api/index.ts b/web/src/app/infra/entities/api/index.ts index c0e79fc0..ce5043ea 100644 --- a/web/src/app/infra/entities/api/index.ts +++ b/web/src/app/infra/entities/api/index.ts @@ -365,6 +365,18 @@ export interface MCPServerExtraArgsSSE { ssereadtimeout: number; } +export interface MCPServerExtraArgsStdio { + command: string; + args: string[]; + env: Record; +} + +export interface MCPServerExtraArgsHttp { + url: string; + headers: Record; + timeout: number; +} + export enum MCPSessionStatus { CONNECTING = 'connecting', CONNECTED = 'connected', @@ -373,21 +385,42 @@ export enum MCPSessionStatus { export interface MCPServerRuntimeInfo { status: MCPSessionStatus; - error_message: string; + error_message?: string; tool_count: number; tools: MCPTool[]; } -export interface MCPServer { - uuid?: string; - name: string; - mode: 'stdio' | 'sse'; - enable: boolean; - extra_args: MCPServerExtraArgsSSE; - runtime_info?: MCPServerRuntimeInfo; - created_at?: string; - updated_at?: string; -} +export type MCPServer = + | { + uuid?: string; + name: string; + mode: 'sse'; + enable: boolean; + extra_args: MCPServerExtraArgsSSE; + runtime_info?: MCPServerRuntimeInfo; + created_at?: string; + updated_at?: string; + } + | { + uuid?: string; + name: string; + mode: 'http'; + enable: boolean; + extra_args: MCPServerExtraArgsHttp; + runtime_info?: MCPServerRuntimeInfo; + created_at?: string; + updated_at?: string; + } + | { + uuid?: string; + name: string; + mode: 'stdio'; + enable: boolean; + extra_args: MCPServerExtraArgsStdio; + runtime_info?: MCPServerRuntimeInfo; + created_at?: string; + updated_at?: string; + }; export interface MCPTool { name: string; diff --git a/web/src/i18n/locales/en-US.ts b/web/src/i18n/locales/en-US.ts index 84bd0fca..cf79c55a 100644 --- a/web/src/i18n/locales/en-US.ts +++ b/web/src/i18n/locales/en-US.ts @@ -467,8 +467,10 @@ const enUS = { getServerListError: 'Failed to get MCP server list: ', serverName: 'Server Name', serverMode: 'Connection Mode', + selectMode: 'Select Mode', stdio: 'Stdio Mode', sse: 'SSE Mode', + http: 'HTTP Mode', noServerInstalled: 'No MCP servers configured', serverNameRequired: 'Server name cannot be empty', commandRequired: 'Command cannot be empty', diff --git a/web/src/i18n/locales/ja-JP.ts b/web/src/i18n/locales/ja-JP.ts index 16f2f386..b812d937 100644 --- a/web/src/i18n/locales/ja-JP.ts +++ b/web/src/i18n/locales/ja-JP.ts @@ -474,6 +474,8 @@ const jaJP = { serverMode: '接続モード', stdio: 'Stdioモード', sse: 'SSEモード', + http: 'HTTPモード', + selectMode: '接続モードを選択', noServerInstalled: 'MCPサーバーが設定されていません', serverNameRequired: 'サーバー名は必須です', commandRequired: 'コマンドは必須です', diff --git a/web/src/i18n/locales/zh-Hans.ts b/web/src/i18n/locales/zh-Hans.ts index 445ae532..473a8401 100644 --- a/web/src/i18n/locales/zh-Hans.ts +++ b/web/src/i18n/locales/zh-Hans.ts @@ -446,8 +446,10 @@ const zhHans = { getServerListError: '获取 MCP 服务器列表失败：', serverName: '服务器名称', serverMode: '连接模式', + selectMode: '选择模式', stdio: 'Stdio模式', sse: 'SSE模式', + http: 'HTTP模式', noServerInstalled: '暂未配置任何 MCP 服务器', serverNameRequired: '服务器名称不能为空', commandRequired: '命令不能为空', diff --git a/web/src/i18n/locales/zh-Hant.ts b/web/src/i18n/locales/zh-Hant.ts index b8369fae..f4cf40b9 100644 --- a/web/src/i18n/locales/zh-Hant.ts +++ b/web/src/i18n/locales/zh-Hant.ts @@ -445,6 +445,8 @@ const zhHant = { serverMode: '連接模式', stdio: 'Stdio模式', sse: 'SSE模式', + selectMode: '選擇連接模式', + http: 'HTTP模式', noServerInstalled: '暫未設定任何MCP伺服器', serverNameRequired: '伺服器名稱不能為空', commandRequired: '命令不能為空',