Strata 是否解决了现阶段MCP的痛点？

本文基于仓库开源实现（https://github.com/Klavis-AI/klavis/tree/main/open-strata），分析 Strata 在大规模集成场景下针对 MCP 痛点所采用的工程设计。

先说结论：如果你的agent需要访问几十上百个mcp服务器，那么Strata有限度的解决了你的问题。
如果你并没有这个需求，那么无需使用。

工具爆炸：多个 MCP 服务器，每个暴露几十上百个工具，LLM 上下文被工具清单淹没，推理退化。
发现困难：缺少统一的按意图发现与筛选机制，模型需要盲猜应该调用哪个服务器的哪个工具。
传输异构：不同服务器使用 stdio、HTTP、SSE，多协议管理复杂。
配置分散：各服务器的启停、认证、Header/Env 等配置分散，缺少热更新与统一观察。
认证繁琐：OAuth 等交互式认证流程不易自动化集成到工具调用链。

统一路由：Strata 自身就是一个 MCP Server，向上游（IDE/代理）只暴露少量“元工具”，再在内部路由到具体 MCP 服务器。
渐进式暴露：先“发现与搜索”，后“细化与执行”，减少一次性向模型暴露全部工具的必要性。
多传输与会话统一：抽象 Transport，同时支持 stdio、HTTP、SSE，复用统一的会话生命周期。
配置中心与热更新：本地 JSON 配置（MCP 格式），文件监控热同步，自动连接/断开。
内置搜索：BM25+ 加权字段检索，支持对多服务器的工具进行统一语义检索。

Strata 仅向上游暴露少量“元工具”，例如（open-strata/src/strata/tools.py:22）：

TOOL_DISCOVER_SERVER_ACTIONS = “discover_server_actions”
TOOL_GET_ACTION_DETAILS = “get_action_details”
TOOL_EXECUTE_ACTION = “execute_action”

python

Server 侧挂载统一处理器（open-strata/src/strata/server.py:81）：

@server.list_tools()
async def list_tools() -> list[types.Tool]:
    user_available_servers = list(client_manager.active_clients.keys())
    return get_tool_definitions(user_available_servers)

@server.call_tool(validate_input=False)
async def call_tool(name: str, arguments: dict) -> list[types.ContentBlock]:
    return await execute_tool(name, arguments, client_manager)

python

“发现”阶段通过 BM25+ 搜索筛选（open-strata/src/strata/tools.py:173）：

searcher = UniversalToolSearcher({server_name: tools})
search_results = searcher.search(user_query, max_results=50)

python

执行阶段精准路由到底层服务器（open-strata/src/strata/tools.py:262）：

client = client_manager.get_client(server_name)
return await client.call_tool(action_name, action_params)

python

抽象基类（open-strata/src/strata/mcp_proxy/transport/base.py:13）：

class Transport(ABC):
    async def _get_streams(self, exit_stack): ...
    async def initialize(self):
        streams = await self._get_streams(self._exit_stack)
        self._session = await self._exit_stack.enter_async_context(
            ClientSession(streams[0], streams[1])
        )
        await self._session.initialize()

python

HTTP/SSE 选择与 OAuth 集成（open-strata/src/strata/mcp_proxy/transport/http.py:59）：

if self.mode == “sse”:
    return await exit_stack.enter_async_context(
        sse_client(self.url, headers=self.headers, auth=create_oauth_provider(self.server_name, self.url))
    )
else:
    return await exit_stack.enter_async_context(
        streamablehttp_client(self.url, headers=self.headers, auth=create_oauth_provider(self.server_name, self.url))
    )

python

Stdio 传输（open-strata/src/strata/mcp_proxy/transport/stdio.py:63）：

server_params = StdioServerParameters(command=self.command, args=self.args, env=self.env)
return await exit_stack.enter_async_context(stdio_client(server_params))

python

配置模型与加载（open-strata/src/strata/config.py:13）：

@dataclass
class MCPServerConfig:
    name: str
    type: str = “stdio”  # stdio/sse/http
    command: Optional[str] = None
    url: Optional[str] = None
    headers: Dict[str, str] = field(default_factory=dict)
    env: Dict[str, str] = field(default_factory=dict)
    enabled: bool = True
    auth: str = “”

python

文件监听与回调（open-strata/src/strata/config.py:288）：

async for _ in awatch(str(self.config_path.parent)):
    self.servers.clear(); self.load(); on_changed(dict(self.servers))

python

统一管理器自动连断（open-strata/src/strata/mcp_client_manager.py:96）：

if server.type in [”sse”,”http”]:
    transport = HTTPTransport(server_name=server.name, url=server.url, mode=server.type, headers=server.headers, auth=server.auth)
else:
    transport = StdioTransport(command=server.command, args=server.args, env=server.env)
client = MCPClient(transport); await client.connect()

python

Strata 作为 MCP Server（open-strata/src/strata/server.py:139 / open-strata/src/strata/server.py:187）：

sse = SseServerTransport(”/messages/”)
session_manager = StreamableHTTPSessionManager(app=app, json_response=json_response, stateless=True)
starlette_app = Starlette(routes=[
  Route(”/sse”, handle_sse, methods=[”GET”]),
  Mount(”/messages/”, app=sse.handle_post_message),
  Mount(”/mcp”, app=handle_streamable_http),
], lifespan=lifespan)

python

由此可见，上游可以使用标准的 SSE 或 Streamable HTTP 与 Strata 通信，随后由 Strata 在内部路由到底层服务器。

对上游接口更小：仅暴露“发现/细化/执行”三类元工具；
工程复杂度更可控：多传输、认证、配置热更、搜索、监控收敛到一处；
规模下的稳定性更易维护：减少一次性暴露大量工具引发的上下文负担与决策干扰。

Discussion about this post