feat(mcp): support mcp resources

src/langbot/pkg/api/http/controller/groups/resources/mcp.py

@@ -66,3 +66,25 @@ class MCPRouterGroup(group.RouterGroup):
             server_data = await quart.request.json
             task_id = await self.ap.mcp_service.test_mcp_server(server_name=server_name, server_data=server_data)
             return self.success(data={'task_id': task_id})
+
+        @self.route('/servers/<server_name>/resources', methods=['GET'], auth_type=group.AuthType.USER_TOKEN)
+        async def _(server_name: str) -> str:
+            """Get resources from an MCP server"""
+            try:
+                resources = await self.ap.mcp_service.get_mcp_server_resources(server_name)
+                return self.success(data={'resources': resources})
+            except Exception as e:
+                return self.http_status(500, -1, f'Failed to get resources: {str(e)}')
+
+        @self.route('/servers/<server_name>/resources/read', methods=['POST'], auth_type=group.AuthType.USER_TOKEN)
+        async def _(server_name: str) -> str:
+            """Read a resource from an MCP server"""
+            data = await quart.request.json
+            uri = data.get('uri')
+            if not uri:
+                return self.http_status(400, -1, 'URI is required')
+            try:
+                contents = await self.ap.mcp_service.read_mcp_server_resource(server_name, uri)
+                return self.success(data={'contents': contents})
+            except Exception as e:
+                return self.http_status(500, -1, f'Failed to read resource: {str(e)}')

src/langbot/pkg/api/http/service/mcp.py

@@ -136,6 +136,14 @@ class MCPService:
             if server_name in self.ap.tool_mgr.mcp_tool_loader.sessions:
                 await self.ap.tool_mgr.mcp_tool_loader.remove_mcp_server(server_name)
 
+    async def get_mcp_server_resources(self, server_name: str) -> list[dict]:
+        """Get resources from a specific MCP server."""
+        return await self.ap.tool_mgr.mcp_tool_loader.get_resources(server_name)
+
+    async def read_mcp_server_resource(self, server_name: str, uri: str) -> list[dict]:
+        """Read a resource from a specific MCP server."""
+        return await self.ap.tool_mgr.mcp_tool_loader.read_resource(server_name, uri)
+
     async def test_mcp_server(self, server_name: str, server_data: dict) -> int:
         """测试 MCP 服务器连接并返回任务 ID"""
 

src/langbot/pkg/provider/tools/loaders/mcp.py

@@ -1,6 +1,8 @@
 from __future__ import annotations
 
+import base64
 import enum
+import json
 import typing
 from contextlib import AsyncExitStack
 import traceback
@@ -10,10 +12,11 @@ import asyncio
 import httpx
 
 import uuid as uuid_module
-from mcp import ClientSession, StdioServerParameters
+from mcp import ClientSession, StdioServerParameters, types as mcp_types
 from mcp.client.stdio import stdio_client
 from mcp.client.sse import sse_client
 from mcp.client.streamable_http import streamable_http_client
+from pydantic import AnyUrl
 
 from .. import loader
 from ....core import app
@@ -22,6 +25,43 @@ import langbot_plugin.api.entities.builtin.provider.message as provider_message
 from ....entity.persistence import mcp as persistence_mcp
 from .mcp_stdio import BoxStdioSessionRuntime, MCPServerBoxConfig, MCPSessionErrorPhase  # noqa: F401
 
+# Synthesized LLM tools for MCP resources (not from server tools/list).
+# Dispatched in MCPLoader.invoke_tool; placeholder func on LLMTool is never used.
+# Prefixed with langbot_ to avoid clashing with MCP server tool names.
+MCP_TOOL_LIST_RESOURCES = 'langbot_mcp_list_resources'
+MCP_TOOL_READ_RESOURCE = 'langbot_mcp_read_resource'
+
+MCP_LIST_RESOURCES_SCHEMA: dict[str, typing.Any] = {
+    'type': 'object',
+    'properties': {
+        'server_name': {
+            'type': 'string',
+            'description': 'MCP server name as configured in LangBot (see admin / pipeline bindings).',
+        }
+    },
+    'required': ['server_name'],
+}
+
+MCP_READ_RESOURCE_SCHEMA: dict[str, typing.Any] = {
+    'type': 'object',
+    'properties': {
+        'server_name': {
+            'type': 'string',
+            'description': 'MCP server name as configured in LangBot.',
+        },
+        'uri': {
+            'type': 'string',
+            'description': 'Resource URI from langbot_mcp_list_resources output, or from MCP documentation.',
+        },
+    },
+    'required': ['server_name', 'uri'],
+}
+
+
+async def _mcp_resource_tool_placeholder(**kwargs: typing.Any) -> list[provider_message.ContentElement]:
+    """LLMTool requires a func; real execution goes through MCPLoader.invoke_tool."""
+    raise RuntimeError('MCP resource tool execution must be routed through MCPLoader.invoke_tool')
+
 
 class MCPSessionStatus(enum.Enum):
     CONNECTING = 'connecting'
@@ -46,6 +86,8 @@ class RuntimeMCPSession:
 
     functions: list[resource_tool.LLMTool] = []
 
+    resources: list[dict] = []
+
     enable: bool
 
     # connected: bool
@@ -82,6 +124,7 @@ class RuntimeMCPSession:
 
         self.exit_stack = AsyncExitStack()
         self.functions = []
+        self.resources = []
 
         self.status = MCPSessionStatus.CONNECTING
 
@@ -253,6 +296,7 @@ class RuntimeMCPSession:
                     await self.exit_stack.aclose()
                     self.exit_stack = AsyncExitStack()
                 self.functions.clear()
+                self.resources.clear()
                 self.session = None
             except Exception as e:
                 self.ap.logger.error(f'Error cleaning up MCP session {self.server_name}: {e}\n{traceback.format_exc()}')
@@ -348,6 +392,7 @@ class RuntimeMCPSession:
             return
 
         self.functions.clear()
+        self.resources.clear()
 
         tools = await self.session.list_tools()
 
@@ -374,8 +419,11 @@ class RuntimeMCPSession:
                     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
+                        if isinstance(content.resource, mcp_types.TextResourceContents):
+                            result_contents.append(provider_message.ContentElement.from_text(content.resource.text))
+                        elif isinstance(content.resource, mcp_types.BlobResourceContents):
+                            decoded = base64.b64decode(content.resource.blob)
+                            result_contents.append(provider_message.ContentElement.from_text(decoded.decode('utf-8', errors='replace')))
 
                 return result_contents
 
@@ -391,9 +439,49 @@ class RuntimeMCPSession:
                 )
             )
 
+        try:
+            resources_result = await self.session.list_resources()
+            for resource in resources_result.resources:
+                self.resources.append({
+                    'uri': str(resource.uri),
+                    'name': resource.name,
+                    'description': resource.description or '',
+                    'mime_type': resource.mimeType or '',
+                })
+            self.ap.logger.debug(f'Refresh MCP resources: {len(self.resources)} resources found')
+        except Exception as e:
+            self.ap.logger.debug(f'MCP server {self.server_name} does not support resources or failed to list: {e}')
+
     def get_tools(self) -> list[resource_tool.LLMTool]:
         return self.functions
 
+    def get_resources(self) -> list[dict]:
+        return self.resources
+
+    async def read_resource(self, uri: str) -> list[dict]:
+        """Read a resource by URI and return its contents."""
+        if not self.session:
+            raise Exception('MCP session is not connected')
+
+        result = await self.session.read_resource(AnyUrl(uri))
+        contents = []
+        for content in result.contents:
+            if isinstance(content, mcp_types.TextResourceContents):
+                contents.append({
+                    'uri': str(content.uri),
+                    'mime_type': content.mimeType or '',
+                    'type': 'text',
+                    'text': content.text,
+                })
+            elif isinstance(content, mcp_types.BlobResourceContents):
+                contents.append({
+                    'uri': str(content.uri),
+                    'mime_type': content.mimeType or '',
+                    'type': 'blob',
+                    'blob': content.blob,
+                })
+        return contents
+
     def get_runtime_info_dict(self) -> dict:
         info = {
             'status': self.status.value,
@@ -409,6 +497,8 @@ class RuntimeMCPSession:
                 }
                 for tool in self.get_tools()
             ],
+            'resource_count': len(self.get_resources()),
+            'resources': self.get_resources(),
         }
         if self._uses_box_stdio():
             info['box_session_id'] = self._build_box_session_id()
@@ -575,8 +665,130 @@ class MCPLoader(loader.ToolLoader):
 
         return session
 
+    @staticmethod
+    def _get_bound_mcp_from_query(query: pipeline_query.Query) -> list[str] | None:
+        v = getattr(query, 'variables', None) or {}
+        return v.get('_pipeline_bound_mcp_servers', None)
+
+    def _eligible_sessions_for_bound(self, bound_mcp_servers: list[str] | None) -> list[RuntimeMCPSession]:
+        out: list[RuntimeMCPSession] = []
+        for session in self.sessions.values():
+            if not session.enable:
+                continue
+            if session.status != MCPSessionStatus.CONNECTED:
+                continue
+            if session.session is None:
+                continue
+            if bound_mcp_servers is not None and session.server_uuid not in bound_mcp_servers:
+                continue
+            out.append(session)
+        return out
+
+    @staticmethod
+    def _mcp_synthetic_resource_tools() -> list[resource_tool.LLMTool]:
+        return [
+            resource_tool.LLMTool(
+                name=MCP_TOOL_LIST_RESOURCES,
+                human_desc='List MCP resource URIs for a server (MCP resources/list).',
+                description=(
+                    'Lists static resources (URI, name, description, mime type) exposed by the MCP server. '
+                    'Call langbot_mcp_read_resource with a URI to fetch content. '
+                    'Use the server name from LangBot pipeline MCP bindings or admin configuration.'
+                ),
+                parameters=MCP_LIST_RESOURCES_SCHEMA,
+                func=_mcp_resource_tool_placeholder,
+            ),
+            resource_tool.LLMTool(
+                name=MCP_TOOL_READ_RESOURCE,
+                human_desc='Read a single MCP resource by URI (MCP resources/read).',
+                description=(
+                    'Fetches the body of a resource. Discover URIs via langbot_mcp_list_resources or MCP docs. '
+                    'The businessId, env, and other parameters in downstream tools must match the selected environment.'
+                ),
+                parameters=MCP_READ_RESOURCE_SCHEMA,
+                func=_mcp_resource_tool_placeholder,
+            ),
+        ]
+
+    async def _invoke_mcp_list_resources(self, parameters: dict, query: pipeline_query.Query) -> typing.Any:
+        server_name = parameters.get('server_name') if parameters else None
+        if not server_name or not isinstance(server_name, str):
+            return [provider_message.ContentElement.from_text('Error: "server_name" (string) is required.')]
+
+        bound = self._get_bound_mcp_from_query(query)
+        allowed = {s.server_name for s in self._eligible_sessions_for_bound(bound)}
+        if server_name not in allowed:
+            return [
+                provider_message.ContentElement.from_text(
+                    f'Error: MCP server {server_name!r} is not available for this query. '
+                    f'Allowed server names: {sorted(allowed)}. '
+                    'Check pipeline MCP server bindings and that the server is connected.'
+                )
+            ]
+
+        session = self.get_session(server_name)
+        if session is None or session.status != MCPSessionStatus.CONNECTED:
+            return [provider_message.ContentElement.from_text(f'Error: MCP server not connected: {server_name!r}')]
+
+        data = session.get_resources()
+        body = {
+            'server_name': server_name,
+            'resource_count': len(data),
+            'resources': data,
+        }
+        return [provider_message.ContentElement.from_text(json.dumps(body, ensure_ascii=False, indent=2))]
+
+    async def _invoke_mcp_read_resource(self, parameters: dict, query: pipeline_query.Query) -> typing.Any:
+        server_name = parameters.get('server_name') if parameters else None
+        uri = parameters.get('uri') if parameters else None
+        if not server_name or not isinstance(server_name, str):
+            return [provider_message.ContentElement.from_text('Error: "server_name" (string) is required.')]
+        if not uri or not isinstance(uri, str):
+            return [provider_message.ContentElement.from_text('Error: "uri" (string) is required.')]
+
+        bound = self._get_bound_mcp_from_query(query)
+        allowed = {s.server_name for s in self._eligible_sessions_for_bound(bound)}
+        if server_name not in allowed:
+            return [
+                provider_message.ContentElement.from_text(
+                    f'Error: MCP server {server_name!r} is not available for this query. '
+                    f'Allowed server names: {sorted(allowed)}.'
+                )
+            ]
+
+        session = self.get_session(server_name)
+        if session is None or session.status != MCPSessionStatus.CONNECTED:
+            return [provider_message.ContentElement.from_text(f'Error: MCP server not connected: {server_name!r}')]
+
+        try:
+            parts = await session.read_resource(uri)
+        except Exception as e:
+            self.ap.logger.error(f'read_resource {uri!r} on {server_name}: {e}\n{traceback.format_exc()}')
+            return [provider_message.ContentElement.from_text(f'Error reading resource: {e!s}')]
+
+        out_chunks: list[str] = []
+        for item in parts:
+            if not isinstance(item, dict):
+                continue
+            t = item.get('type', '')
+            if t == 'text' and 'text' in item:
+                out_chunks.append(typing.cast(str, item['text']))
+            elif t == 'blob' and 'blob' in item:
+                try:
+                    raw = base64.b64decode(typing.cast(str, item['blob']))
+                    out_chunks.append(raw.decode('utf-8', errors='replace'))
+                except Exception as be:
+                    out_chunks.append(f'[Binary decode error: {be}]')
+        if not out_chunks:
+            return [
+                provider_message.ContentElement.from_text(
+                    json.dumps({'uri': uri, 'contents': parts}, ensure_ascii=False, indent=2)
+                )
+            ]
+        return [provider_message.ContentElement.from_text('\n\n'.join(out_chunks))]
+
     async def get_tools(self, bound_mcp_servers: list[str] | None = None) -> list[resource_tool.LLMTool]:
-        all_functions = []
+        all_functions: list[resource_tool.LLMTool] = []
 
         for session in self.sessions.values():
             # If bound_mcp_servers is specified, only include tools from those servers
@@ -587,12 +799,17 @@ class MCPLoader(loader.ToolLoader):
                 # If no bound servers specified, include all tools
                 all_functions.extend(session.get_tools())
 
+        if self._eligible_sessions_for_bound(bound_mcp_servers):
+            all_functions.extend(self._mcp_synthetic_resource_tools())
+
         self._last_listed_functions = all_functions
 
         return all_functions
 
     async def has_tool(self, name: str) -> bool:
         """检查工具是否存在"""
+        if name in (MCP_TOOL_LIST_RESOURCES, MCP_TOOL_READ_RESOURCE):
+            return bool(self._eligible_sessions_for_bound(None))
         for session in self.sessions.values():
             for function in session.get_tools():
                 if function.name == name:
@@ -608,6 +825,11 @@ class MCPLoader(loader.ToolLoader):
 
     async def invoke_tool(self, name: str, parameters: dict, query: pipeline_query.Query) -> typing.Any:
         """执行工具调用"""
+        if name == MCP_TOOL_LIST_RESOURCES:
+            return await self._invoke_mcp_list_resources(parameters, query)
+        if name == MCP_TOOL_READ_RESOURCE:
+            return await self._invoke_mcp_read_resource(parameters, query)
+
         for session in self.sessions.values():
             for function in session.get_tools():
                 if function.name == name:
@@ -622,6 +844,20 @@ class MCPLoader(loader.ToolLoader):
 
         raise ValueError(f'Tool not found: {name}')
 
+    async def get_resources(self, server_name: str) -> list[dict]:
+        """Get resources from a specific MCP server."""
+        session = self.get_session(server_name)
+        if session is None:
+            raise ValueError(f'MCP server not found: {server_name}')
+        return session.get_resources()
+
+    async def read_resource(self, server_name: str, uri: str) -> list[dict]:
+        """Read a resource from a specific MCP server."""
+        session = self.get_session(server_name)
+        if session is None:
+            raise ValueError(f'MCP server not found: {server_name}')
+        return await session.read_resource(uri)
+
     async def remove_mcp_server(self, server_name: str):
         """移除 MCP 服务器"""
         if server_name not in self.sessions: