콘텐츠로 이동

spakky-mcp

spakky-mcp는 외부 MCP server tools를 Spakky Agent run에 연결하는 단방향 adapter입니다.

McpClient가 IAgentRunnerFactory 구현체로 외부 서버 연결 수명주기를 소유합니다. McpRuntimeServerResolver는 configured server 이름과 run-time inline server 선언을 해석하고, descriptor 계층은 발견된 MCP tools를 lazy search/call meta-tools 뒤에 보관합니다.

Public API

MCP adapter plugin joining external server tools to Spakky Agents.

MCP_CALL_TOOL_NAME = 'mcp_call_tool' module-attribute

Model-facing lazy invocation tool for a discovered MCP tool.

MCP_SEARCH_TOOLS_NAME = 'mcp_search_tools' module-attribute

Model-facing lazy discovery tool for the current run's MCP toolset.

MCP_METADATA_KEY = 'mcp' module-attribute

RunAgentInput.metadata key carrying runtime MCP connection selectors.

MCP_SERVERS_METADATA_KEY = 'servers' module-attribute

Nested metadata key carrying server names or inline server declarations.

PLUGIN_NAME = Plugin(name='spakky-mcp') module-attribute

Plugin identifier for the MCP adapter package.

MCPClient = McpClient module-attribute

Uppercase-acronym alias for :class:McpClient.

MCPConfig = McpConfig module-attribute

Uppercase-acronym alias for :class:McpConfig.

MCPHttpClientProvider = McpHttpClientProvider module-attribute

Uppercase-acronym alias for :class:McpHttpClientProvider.

MCPRuntimeServerResolver = McpRuntimeServerResolver module-attribute

Uppercase-acronym alias for :class:McpRuntimeServerResolver.

MCPServerAuthConfig = McpServerAuthConfig module-attribute

Uppercase-acronym alias for :class:McpServerAuthConfig.

MCPServerConfig = McpServerConfig module-attribute

Uppercase-acronym alias for :class:McpServerConfig.

MCPTransport = McpTransport module-attribute

Uppercase-acronym alias for :class:McpTransport.

IMcpHttpClientProvider

Bases: ABC

Factory for authenticated HTTP clients used by streamable_http MCP servers.

open_client(server) abstractmethod

Open an optional HTTP client for one server connection.

Source code in plugins/spakky-mcp/src/spakky/plugins/mcp/auth.py 
28
29
30
31
32
33
34
@abstractmethod
def open_client(
    self,
    server: McpServerConfig,
) -> AbstractAsyncContextManager[httpx.AsyncClient | None]:
    """Open an optional HTTP client for one server connection."""
    ...

McpHttpClientProvider

Bases: IMcpHttpClientProvider

Default declarative HTTP auth provider for remote MCP servers.

open_client(server) async

Yield a configured HTTP client when the server declares auth headers.

Source code in plugins/spakky-mcp/src/spakky/plugins/mcp/auth.py 
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
@asynccontextmanager
async def open_client(
    self,
    server: McpServerConfig,
) -> AsyncGenerator[httpx.AsyncClient | None, None]:
    """Yield a configured HTTP client when the server declares auth headers."""
    if server.transport is not McpTransport.STREAMABLE_HTTP:
        yield None
        return
    headers = await resolve_http_auth_headers(server.auth)
    if not headers:
        yield None
        return
    async with httpx.AsyncClient(headers=headers) as client:
        yield client

McpClient(config, http_client_provider=None, runtime_server_resolver=None, runner_factory=None)

Bases: IAgentRunnerFactory

Runner factory that joins external MCP tools to an agent runner.

Source code in plugins/spakky-mcp/src/spakky/plugins/mcp/client.py 
167
168
169
170
171
172
173
174
175
176
177
178
179
def __init__(
    self,
    config: McpConfig,
    http_client_provider: IMcpHttpClientProvider | None = None,
    runtime_server_resolver: IMcpRuntimeServerResolver | None = None,
    runner_factory: AgentRunnerFactory | None = None,
) -> None:
    self.config = config
    self._runner_factory = runner_factory or AgentRunnerFactory()
    self._http_client_provider = http_client_provider or McpHttpClientProvider()
    self._runtime_server_resolver = (
        runtime_server_resolver or McpRuntimeServerResolver(config)
    )

open_runner(agent_instance, run_input=None) async

Yield a runner whose catalog also carries the external MCP tools.

Source code in plugins/spakky-mcp/src/spakky/plugins/mcp/client.py 
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
@asynccontextmanager
@override
async def open_runner(
    self,
    agent_instance: object,
    run_input: RunAgentInput | None = None,
) -> AsyncGenerator[AgentRunner, None]:
    """Yield a runner whose catalog also carries the external MCP tools."""
    descriptors: list[AgentToolDescriptor] = []
    servers = self._runtime_server_resolver.resolve_servers(
        agent_instance,
        run_input,
    )
    async with AsyncExitStack() as stack:
        for server in servers:
            http_client = await stack.enter_async_context(
                self._http_client_provider.open_client(server)
            )
            _session, server_descriptors = await stack.enter_async_context(
                connect_server(
                    server,
                    self.config.connect_timeout_seconds,
                    http_client,
                )
            )
            descriptors.extend(server_descriptors)
        runner = await stack.enter_async_context(
            self._runner_factory.open_runner(agent_instance, run_input=run_input)
        )
        yield build_mcp_runner(runner, descriptors)

McpOAuthClientAuthMethod

Bases: StrEnum

Client authentication method for OAuth2 client-credentials token requests.

McpOAuthClientCredentialsConfig

Bases: BaseModel

OAuth2 client-credentials declaration for an authenticated MCP server.

McpConfig()

Bases: BaseSettings

Settings declaring the external MCP servers an agent consumes.

Source code in plugins/spakky-mcp/src/spakky/plugins/mcp/config.py 
236
237
def __init__(self) -> None:
    super().__init__()

servers = () class-attribute instance-attribute

External MCP servers whose tools join the agent tool catalog.

connect_timeout_seconds = DEFAULT_MCP_CONNECT_TIMEOUT_SECONDS class-attribute instance-attribute

Timeout budget for establishing an MCP server connection.

server_by_name(name)

Return the declared server with the given name.

Source code in plugins/spakky-mcp/src/spakky/plugins/mcp/config.py 
245
246
247
248
249
250
def server_by_name(self, name: str) -> McpServerConfig:
    """Return the declared server with the given name."""
    for server in self.servers:
        if server.name == name:
            return server
    raise McpServerConfigurationError("MCP server name is not declared")

McpServerAuthConfig

Bases: BaseModel

HTTP authentication declaration for a remote streamable_http MCP server.

McpServerConfig

Bases: BaseModel

Declaration of one external MCP server the agent consumes tools from.

McpTransport

Bases: StrEnum

Transport an external MCP server is reached over.

LazyMcpToolset

Sentinel owner type for MCP lazy search/call descriptors.

AbstractMcpError

Bases: AbstractSpakkyFrameworkError, ABC

Base class for MCP adapter errors.

McpCatalogMergeError

Bases: AbstractMcpError

Raised when an external MCP tool collides with an existing catalog tool.

McpResponseError

Bases: AbstractMcpError

Raised when an MCP tool result cannot be mapped to a JSON value.

McpServerConfigurationError

Bases: AbstractMcpError

Raised when an external MCP server declaration is invalid.

McpToolDiscoveryError

Bases: AbstractMcpError

Raised when tool discovery against an MCP server fails.

McpToolInvocationError

Bases: AbstractMcpError

Raised when an external MCP tool call fails or reports an error result.

McpTransportError

Bases: AbstractMcpError

Raised when an external MCP server connection cannot be established.

IMcpRuntimeServerResolver

Bases: ABC

Resolve MCP servers to join for one Agent run.

resolve_servers(agent_instance, run_input) abstractmethod

Return the MCP server configs selected for this run.

Source code in plugins/spakky-mcp/src/spakky/plugins/mcp/runtime.py 
26
27
28
29
30
31
32
33
@abstractmethod
def resolve_servers(
    self,
    agent_instance: object,
    run_input: RunAgentInput | None,
) -> tuple[McpServerConfig, ...]:
    """Return the MCP server configs selected for this run."""
    ...

McpRuntimeServerResolver(config)

Bases: IMcpRuntimeServerResolver

Default resolver using configured servers plus RunAgentInput metadata.

Source code in plugins/spakky-mcp/src/spakky/plugins/mcp/runtime.py 
40
41
def __init__(self, config: McpConfig) -> None:
    self._config = config

resolve_servers(agent_instance, run_input)

Resolve runtime metadata or all configured servers.

Source code in plugins/spakky-mcp/src/spakky/plugins/mcp/runtime.py 
43
44
45
46
47
48
49
50
51
52
53
54
55
def resolve_servers(
    self,
    agent_instance: object,
    run_input: RunAgentInput | None,
) -> tuple[McpServerConfig, ...]:
    """Resolve runtime metadata or all configured servers."""
    _ = agent_instance
    runtime_servers = _runtime_servers_from_input(run_input)
    if runtime_servers is None:
        return validate_unique_server_names(self._config.servers)
    return validate_unique_server_names(
        tuple(self._runtime_server(item) for item in runtime_servers)
    )

resolve_http_auth_headers(auth, env=environ) async

Return HTTP headers for an authenticated streamable_http MCP connection.

Source code in plugins/spakky-mcp/src/spakky/plugins/mcp/auth.py 
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
async def resolve_http_auth_headers(
    auth: McpServerAuthConfig,
    env: Mapping[str, str] = environ,
) -> dict[str, str]:
    """Return HTTP headers for an authenticated streamable_http MCP connection."""
    headers = dict(auth.headers)
    if auth.bearer_token is not None or auth.bearer_token_env is not None:
        token = _resolve_secret(
            value=auth.bearer_token,
            env_name=auth.bearer_token_env,
            env=env,
            label="MCP bearer token",
        )
        headers["Authorization"] = f"Bearer {token}"
        return headers
    if auth.oauth_client_credentials is not None:
        token = await _fetch_oauth_client_credentials_token(
            auth.oauth_client_credentials,
            env,
        )
        headers["Authorization"] = f"Bearer {token}"
    return headers

build_lazy_mcp_descriptors(external)

Return the two model-visible tools that lazily expose MCP tools.

Source code in plugins/spakky-mcp/src/spakky/plugins/mcp/descriptor.py 
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
def build_lazy_mcp_descriptors(
    external: Sequence[AgentToolDescriptor],
) -> tuple[AgentToolDescriptor, ...]:
    """Return the two model-visible tools that lazily expose MCP tools."""
    if not external:
        return ()
    by_name = {descriptor.schema.name: descriptor for descriptor in external}

    async def search_tools(
        query: str = "",
        limit: int = DEFAULT_MCP_SEARCH_LIMIT,
    ) -> JsonValue:
        """Search the MCP tools available to this run."""
        if limit <= 0:
            raise McpToolInvocationError("MCP search limit must be positive")
        matches = _filter_tool_summaries(external, query)
        return {
            "tools": matches[:limit],
            "count": min(len(matches), limit),
            "total": len(matches),
        }

    async def call_tool(tool_name: str, arguments: JsonObject) -> JsonValue:
        """Call one MCP tool by the name returned from mcp_search_tools."""
        if not tool_name.strip():
            raise McpToolInvocationError("MCP tool name cannot be blank")
        descriptor = by_name.get(tool_name)
        if descriptor is None:
            raise McpToolInvocationError("MCP tool name is not available")
        if not isinstance(arguments, dict):
            raise McpToolInvocationError("MCP tool arguments must be an object")
        bound = descriptor.bind_invocation(arguments)
        result = descriptor.callable(*bound.args, **bound.kwargs)
        if isawaitable(result):
            return cast(JsonValue, await result)
        return cast(JsonValue, result)

    return (
        _lazy_descriptor(
            name=MCP_SEARCH_TOOLS_NAME,
            callable_=search_tools,
            description=(
                "Search the MCP tools connected to this run. Use this before "
                "calling an MCP tool so the full external tool catalog does not "
                "need to be loaded into the initial model context."
            ),
            input_schema={
                "type": "object",
                "title": f"{MCP_SEARCH_TOOLS_NAME}.input",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "Tool name, capability, or domain to search.",
                    },
                    "limit": {
                        "type": "integer",
                        "minimum": 1,
                        "default": DEFAULT_MCP_SEARCH_LIMIT,
                    },
                },
                "additionalProperties": False,
            },
            effects=ToolEffects.read_only(),
            approval=ToolApprovalRequirement.NOT_REQUIRED,
        ),
        _lazy_descriptor(
            name=MCP_CALL_TOOL_NAME,
            callable_=call_tool,
            description=(
                "Call one MCP tool returned by mcp_search_tools. Pass the "
                "tool_name exactly as returned and put the target tool payload "
                "inside arguments."
            ),
            input_schema={
                "type": "object",
                "title": f"{MCP_CALL_TOOL_NAME}.input",
                "properties": {
                    "tool_name": {
                        "type": "string",
                        "description": "Exact MCP tool name returned by search.",
                    },
                    "arguments": {
                        "type": "object",
                        "description": "JSON object forwarded to the MCP tool.",
                        "additionalProperties": True,
                    },
                },
                "required": ["tool_name", "arguments"],
                "additionalProperties": False,
            },
            effects=ToolEffects.external_side_effect(),
            approval=ToolApprovalRequirement.DERIVED,
            descriptor_type=LazyMcpCallToolDescriptor,
        ),
    )

build_mcp_runner(runner, external)

Build a runner whose catalog carries lazy MCP search/call tools.

The caller supplies an already-open native runner so any request-scoped model resolver or durable-port assembly has already happened. The agent Pod metadata is a shared singleton, so its catalog is augmented on a copy.copy. The actual external MCP descriptors stay hidden behind lazy meta-tools so a large MCP server does not flood the model request with every tool schema.

Source code in plugins/spakky-mcp/src/spakky/plugins/mcp/descriptor.py 
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
def build_mcp_runner(
    runner: AgentRunner,
    external: Sequence[AgentToolDescriptor],
) -> AgentRunner:
    """Build a runner whose catalog carries lazy MCP search/call tools.

    The caller supplies an already-open native runner so any request-scoped model
    resolver or durable-port assembly has already happened. The agent Pod
    metadata is a shared singleton, so its catalog is augmented on a ``copy.copy``.
    The actual external MCP descriptors stay hidden behind lazy meta-tools so a
    large MCP server does not flood the model request with every tool schema.
    """
    merged = merge_external_catalog(
        runner.agent.tool_catalog,
        build_lazy_mcp_descriptors(external),
    )
    augmented_agent = copy.copy(runner.agent)
    augmented_agent.tool_catalog = merged
    return dataclasses.replace(runner, agent=augmented_agent)

설정

Configuration for connecting external MCP servers to Spakky Agent runs.

McpTransport

Bases: StrEnum

Transport an external MCP server is reached over.

McpOAuthClientAuthMethod

Bases: StrEnum

Client authentication method for OAuth2 client-credentials token requests.

McpOAuthClientCredentialsConfig

Bases: BaseModel

OAuth2 client-credentials declaration for an authenticated MCP server.

McpServerAuthConfig

Bases: BaseModel

HTTP authentication declaration for a remote streamable_http MCP server.

McpServerConfig

Bases: BaseModel

Declaration of one external MCP server the agent consumes tools from.

McpConfig()

Bases: BaseSettings

Settings declaring the external MCP servers an agent consumes.

Source code in plugins/spakky-mcp/src/spakky/plugins/mcp/config.py 
236
237
def __init__(self) -> None:
    super().__init__()

servers = () class-attribute instance-attribute

External MCP servers whose tools join the agent tool catalog.

connect_timeout_seconds = DEFAULT_MCP_CONNECT_TIMEOUT_SECONDS class-attribute instance-attribute

Timeout budget for establishing an MCP server connection.

server_by_name(name)

Return the declared server with the given name.

Source code in plugins/spakky-mcp/src/spakky/plugins/mcp/config.py 
245
246
247
248
249
250
def server_by_name(self, name: str) -> McpServerConfig:
    """Return the declared server with the given name."""
    for server in self.servers:
        if server.name == name:
            return server
    raise McpServerConfigurationError("MCP server name is not declared")

validate_unique_server_names(servers)

Reject duplicate MCP server names before runtime selection is ambiguous.

Source code in plugins/spakky-mcp/src/spakky/plugins/mcp/config.py 
208
209
210
211
212
213
214
215
216
217
def validate_unique_server_names(
    servers: tuple[McpServerConfig, ...],
) -> tuple[McpServerConfig, ...]:
    """Reject duplicate MCP server names before runtime selection is ambiguous."""
    seen: set[str] = set()
    for server in servers:
        if server.name in seen:
            raise McpServerConfigurationError("MCP server names must be unique")
        seen.add(server.name)
    return servers

Constants for the spakky-mcp external server adapter.

Client

Connection lifecycle and tool discovery for external MCP servers (issue #416).

An MCP ClientSession is only usable inside its transport context, so the callables this module binds to descriptors close over a live session and stay valid only while the connection is open. McpClient.open_runner keeps every configured server's session open for the duration of the yielded runner, then tears the connections down on exit.

McpClient(config, http_client_provider=None, runtime_server_resolver=None, runner_factory=None)

Bases: IAgentRunnerFactory

Runner factory that joins external MCP tools to an agent runner.

Source code in plugins/spakky-mcp/src/spakky/plugins/mcp/client.py 
167
168
169
170
171
172
173
174
175
176
177
178
179
def __init__(
    self,
    config: McpConfig,
    http_client_provider: IMcpHttpClientProvider | None = None,
    runtime_server_resolver: IMcpRuntimeServerResolver | None = None,
    runner_factory: AgentRunnerFactory | None = None,
) -> None:
    self.config = config
    self._runner_factory = runner_factory or AgentRunnerFactory()
    self._http_client_provider = http_client_provider or McpHttpClientProvider()
    self._runtime_server_resolver = (
        runtime_server_resolver or McpRuntimeServerResolver(config)
    )

open_runner(agent_instance, run_input=None) async

Yield a runner whose catalog also carries the external MCP tools.

Source code in plugins/spakky-mcp/src/spakky/plugins/mcp/client.py 
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
@asynccontextmanager
@override
async def open_runner(
    self,
    agent_instance: object,
    run_input: RunAgentInput | None = None,
) -> AsyncGenerator[AgentRunner, None]:
    """Yield a runner whose catalog also carries the external MCP tools."""
    descriptors: list[AgentToolDescriptor] = []
    servers = self._runtime_server_resolver.resolve_servers(
        agent_instance,
        run_input,
    )
    async with AsyncExitStack() as stack:
        for server in servers:
            http_client = await stack.enter_async_context(
                self._http_client_provider.open_client(server)
            )
            _session, server_descriptors = await stack.enter_async_context(
                connect_server(
                    server,
                    self.config.connect_timeout_seconds,
                    http_client,
                )
            )
            descriptors.extend(server_descriptors)
        runner = await stack.enter_async_context(
            self._runner_factory.open_runner(agent_instance, run_input=run_input)
        )
        yield build_mcp_runner(runner, descriptors)

make_mcp_tool_callable(session, raw_tool_name, call_timeout_seconds)

Bind an owner-less async callable that invokes one external MCP tool.

The callable's only parameter is **arguments: the dispatcher's owner-prefix step skips it (no leading self/cls) and binds the model payload straight to the keyword arguments forwarded to call_tool. The configured per-server timeout bounds each call so a hung external tool cannot block the agent loop indefinitely.

Source code in plugins/spakky-mcp/src/spakky/plugins/mcp/client.py 
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
def make_mcp_tool_callable(
    session: ClientSession,
    raw_tool_name: str,
    call_timeout_seconds: float,
) -> McpToolCallable:
    """Bind an owner-less async callable that invokes one external MCP tool.

    The callable's only parameter is ``**arguments``: the dispatcher's
    owner-prefix step skips it (no leading self/cls) and binds the model payload
    straight to the keyword arguments forwarded to ``call_tool``. The configured
    per-server timeout bounds each call so a hung external tool cannot block the
    agent loop indefinitely.
    """
    read_timeout = timedelta(seconds=call_timeout_seconds)

    async def invoke(**arguments: object) -> JsonValue:
        try:
            result = await session.call_tool(
                raw_tool_name,
                arguments=dict(arguments),
                read_timeout_seconds=read_timeout,
            )
        except McpToolInvocationError:
            raise
        except Exception as e:  # MCP/transport failures surface as a typed error.
            raise McpToolInvocationError from e
        return normalize_call_result(result)

    return invoke

connect_server(server, connect_timeout_seconds, http_client=None) async

Open a server connection and discover its tools as catalog descriptors.

connect_timeout_seconds bounds the initialize handshake (the session read timeout) so an unresponsive server fails fast instead of hanging the connection.

Source code in plugins/spakky-mcp/src/spakky/plugins/mcp/client.py 
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
@asynccontextmanager
async def connect_server(
    server: McpServerConfig,
    connect_timeout_seconds: float,
    http_client: AsyncClient | None = None,
) -> AsyncGenerator[DiscoveredServer, None]:
    """Open a server connection and discover its tools as catalog descriptors.

    ``connect_timeout_seconds`` bounds the ``initialize`` handshake (the session
    read timeout) so an unresponsive server fails fast instead of hanging the
    connection.
    """
    try:
        async with (
            _transport_streams(server, http_client) as (read, write),
            ClientSession(
                read,
                write,
                read_timeout_seconds=timedelta(seconds=connect_timeout_seconds),
            ) as session,
        ):
            await session.initialize()
            descriptors = await _discover_descriptors(session, server)
            yield session, descriptors
    except (McpToolDiscoveryError, McpToolInvocationError):
        raise
    except Exception as e:  # connection/initialize failures surface as transport.
        raise McpTransportError from e

Runtime Resolution

Runtime MCP connection resolution for user/service supplied toolsets.

MCP_METADATA_KEY = 'mcp' module-attribute

RunAgentInput.metadata key carrying runtime MCP connection selectors.

MCP_SERVERS_METADATA_KEY = 'servers' module-attribute

Nested metadata key carrying server names or inline server declarations.

IMcpRuntimeServerResolver

Bases: ABC

Resolve MCP servers to join for one Agent run.

resolve_servers(agent_instance, run_input) abstractmethod

Return the MCP server configs selected for this run.

Source code in plugins/spakky-mcp/src/spakky/plugins/mcp/runtime.py 
26
27
28
29
30
31
32
33
@abstractmethod
def resolve_servers(
    self,
    agent_instance: object,
    run_input: RunAgentInput | None,
) -> tuple[McpServerConfig, ...]:
    """Return the MCP server configs selected for this run."""
    ...

McpRuntimeServerResolver(config)

Bases: IMcpRuntimeServerResolver

Default resolver using configured servers plus RunAgentInput metadata.

Source code in plugins/spakky-mcp/src/spakky/plugins/mcp/runtime.py 
40
41
def __init__(self, config: McpConfig) -> None:
    self._config = config

resolve_servers(agent_instance, run_input)

Resolve runtime metadata or all configured servers.

Source code in plugins/spakky-mcp/src/spakky/plugins/mcp/runtime.py 
43
44
45
46
47
48
49
50
51
52
53
54
55
def resolve_servers(
    self,
    agent_instance: object,
    run_input: RunAgentInput | None,
) -> tuple[McpServerConfig, ...]:
    """Resolve runtime metadata or all configured servers."""
    _ = agent_instance
    runtime_servers = _runtime_servers_from_input(run_input)
    if runtime_servers is None:
        return validate_unique_server_names(self._config.servers)
    return validate_unique_server_names(
        tuple(self._runtime_server(item) for item in runtime_servers)
    )

Descriptor

descriptor 모듈은 McpClient가 발견한 외부 MCP tools를 lazy mcp_search_tools / mcp_call_tool 표면 뒤에 보관하기 위한 내부 정규화 계층입니다. 애플리케이션 코드는 일반적으로 McpClient, McpConfig, IMcpRuntimeServerResolver만 사용합니다. mcp_call_tool approval request는 meta-tool이 아니라 선택된 외부 MCP tool 이름과 arguments를 노출합니다.

Normalize external MCP servers into the agent tool catalog (issue #416).

ADR-0013 §2 keeps core/spakky-agent protocol-neutral and pushes the MCP library dependency into this adapter plugin. Discovered MCP tools are kept in a session-local registry and exposed to the model through two lazy meta-tools: search the MCP toolset, then call one selected tool. This keeps large MCP servers out of the initial model tool list while preserving the same dispatcher path for the final invocation.

MCP_SEARCH_TOOLS_NAME = 'mcp_search_tools' module-attribute

Model-facing lazy discovery tool for the current run's MCP toolset.

MCP_CALL_TOOL_NAME = 'mcp_call_tool' module-attribute

Model-facing lazy invocation tool for a discovered MCP tool.

ExternalMcpTool

Sentinel owner type for catalog descriptors discovered from MCP servers.

LazyMcpToolset

Sentinel owner type for MCP lazy search/call descriptors.

ExternalMcpToolDescriptor(identity, owner, callable, schema, description=None, metadata=AgentToolMetadata()) dataclass

Bases: AgentToolDescriptor

Descriptor that binds the MCP argument object verbatim to its callable.

The core binder (bind_agent_tool_invocation) reserves top-level args and kwargs payload keys for its positional/keyword structured-call form. An external MCP tool may legitimately declare input fields named args or kwargs; routing such a payload through that heuristic would fail to bind or drop the field name. MCP tool inputs are always a flat JSON object, so this descriptor forwards the whole payload as keyword arguments without the structured-call interpretation, preserving every declared field name.

bind_invocation(payload)

Forward the MCP argument object as keyword arguments verbatim.

Source code in plugins/spakky-mcp/src/spakky/plugins/mcp/descriptor.py 
80
81
82
83
@override
def bind_invocation(self, payload: JsonObject) -> AgentToolBoundInvocation:
    """Forward the MCP argument object as keyword arguments verbatim."""
    return AgentToolBoundInvocation(args=(), kwargs=dict(payload))

LazyMcpCallToolDescriptor(identity, owner, callable, schema, description=None, metadata=AgentToolMetadata()) dataclass

Bases: AgentToolDescriptor

Lazy call descriptor that surfaces the selected external tool to HITL.

approval_context(payload)

Expose the target MCP tool name and arguments to approval requests.

Source code in plugins/spakky-mcp/src/spakky/plugins/mcp/descriptor.py 
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
@override
def approval_context(self, payload: JsonObject) -> AgentToolApprovalContext:
    """Expose the target MCP tool name and arguments to approval requests."""
    tool_name = payload.get("tool_name")
    if not isinstance(tool_name, str) or not tool_name.strip():
        return AgentToolApprovalContext()
    arguments = payload.get("arguments")
    safe_arguments = (
        cast(JsonObject, dict(arguments)) if isinstance(arguments, dict) else {}
    )
    return AgentToolApprovalContext(
        prompt=f"Approve MCP tool invocation: {tool_name.strip()}",
        action_ref=_external_tool_action_ref(tool_name.strip()),
        metadata={
            "mcp_meta_tool": MCP_CALL_TOOL_NAME,
            "mcp_tool_name": tool_name.strip(),
            "mcp_arguments": safe_arguments,
        },
    )

prefixed_tool_name(server_name, raw_tool_name)

Return the collision-safe model-facing name for an external tool.

Source code in plugins/spakky-mcp/src/spakky/plugins/mcp/descriptor.py 
110
111
112
def prefixed_tool_name(server_name: str, raw_tool_name: str) -> str:
    """Return the collision-safe model-facing name for an external tool."""
    return f"{server_name}{MCP_TOOL_NAME_SEPARATOR}{raw_tool_name}"

build_external_descriptor(server_name, tool, callable_)

Normalize one discovered MCP tool into a catalog descriptor.

Source code in plugins/spakky-mcp/src/spakky/plugins/mcp/descriptor.py 
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
def build_external_descriptor(
    server_name: str,
    tool: Tool,
    callable_: McpToolCallable,
) -> AgentToolDescriptor:
    """Normalize one discovered MCP tool into a catalog descriptor."""
    name = prefixed_tool_name(server_name, tool.name)
    identity = AgentToolIdentity(
        owner_module=MCP_EXTERNAL_TOOL_OWNER_MODULE,
        owner_qualname=f"{ExternalMcpTool.__qualname__}.{server_name}",
        name=name,
    )
    output_schema = cast(JsonObject, tool.outputSchema) if tool.outputSchema else {}
    schema = AgentToolSchemaHandle(
        name=name,
        input_schema_name=f"{name}.input",
        output_schema_name=f"{name}.output",
        input_schema=_normalize_input_schema(tool.inputSchema),
        output_schema=output_schema,
    )
    return ExternalMcpToolDescriptor(
        identity=identity,
        owner=ExternalMcpTool,
        callable=callable_,
        schema=schema,
        description=tool.description,
        metadata=_external_tool_metadata(),
    )

build_external_descriptors(server_name, tools, callable_factory)

Normalize all discovered tools of one server into catalog descriptors.

Source code in plugins/spakky-mcp/src/spakky/plugins/mcp/descriptor.py 
145
146
147
148
149
150
151
152
153
154
def build_external_descriptors(
    server_name: str,
    tools: Sequence[Tool],
    callable_factory: McpToolCallableFactory,
) -> tuple[AgentToolDescriptor, ...]:
    """Normalize all discovered tools of one server into catalog descriptors."""
    return tuple(
        build_external_descriptor(server_name, tool, callable_factory(tool.name))
        for tool in tools
    )

merge_external_catalog(native, external)

Return a catalog combining native descriptors with external MCP tools.

Source code in plugins/spakky-mcp/src/spakky/plugins/mcp/descriptor.py 
157
158
159
160
161
162
163
164
165
def merge_external_catalog(
    native: AgentToolCatalog,
    external: Sequence[AgentToolDescriptor],
) -> AgentToolCatalog:
    """Return a catalog combining native descriptors with external MCP tools."""
    try:
        return AgentToolCatalog(descriptors=(*native.descriptors, *external))
    except AgentDefinitionError as e:
        raise McpCatalogMergeError from e

build_lazy_mcp_descriptors(external)

Return the two model-visible tools that lazily expose MCP tools.

Source code in plugins/spakky-mcp/src/spakky/plugins/mcp/descriptor.py 
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
def build_lazy_mcp_descriptors(
    external: Sequence[AgentToolDescriptor],
) -> tuple[AgentToolDescriptor, ...]:
    """Return the two model-visible tools that lazily expose MCP tools."""
    if not external:
        return ()
    by_name = {descriptor.schema.name: descriptor for descriptor in external}

    async def search_tools(
        query: str = "",
        limit: int = DEFAULT_MCP_SEARCH_LIMIT,
    ) -> JsonValue:
        """Search the MCP tools available to this run."""
        if limit <= 0:
            raise McpToolInvocationError("MCP search limit must be positive")
        matches = _filter_tool_summaries(external, query)
        return {
            "tools": matches[:limit],
            "count": min(len(matches), limit),
            "total": len(matches),
        }

    async def call_tool(tool_name: str, arguments: JsonObject) -> JsonValue:
        """Call one MCP tool by the name returned from mcp_search_tools."""
        if not tool_name.strip():
            raise McpToolInvocationError("MCP tool name cannot be blank")
        descriptor = by_name.get(tool_name)
        if descriptor is None:
            raise McpToolInvocationError("MCP tool name is not available")
        if not isinstance(arguments, dict):
            raise McpToolInvocationError("MCP tool arguments must be an object")
        bound = descriptor.bind_invocation(arguments)
        result = descriptor.callable(*bound.args, **bound.kwargs)
        if isawaitable(result):
            return cast(JsonValue, await result)
        return cast(JsonValue, result)

    return (
        _lazy_descriptor(
            name=MCP_SEARCH_TOOLS_NAME,
            callable_=search_tools,
            description=(
                "Search the MCP tools connected to this run. Use this before "
                "calling an MCP tool so the full external tool catalog does not "
                "need to be loaded into the initial model context."
            ),
            input_schema={
                "type": "object",
                "title": f"{MCP_SEARCH_TOOLS_NAME}.input",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "Tool name, capability, or domain to search.",
                    },
                    "limit": {
                        "type": "integer",
                        "minimum": 1,
                        "default": DEFAULT_MCP_SEARCH_LIMIT,
                    },
                },
                "additionalProperties": False,
            },
            effects=ToolEffects.read_only(),
            approval=ToolApprovalRequirement.NOT_REQUIRED,
        ),
        _lazy_descriptor(
            name=MCP_CALL_TOOL_NAME,
            callable_=call_tool,
            description=(
                "Call one MCP tool returned by mcp_search_tools. Pass the "
                "tool_name exactly as returned and put the target tool payload "
                "inside arguments."
            ),
            input_schema={
                "type": "object",
                "title": f"{MCP_CALL_TOOL_NAME}.input",
                "properties": {
                    "tool_name": {
                        "type": "string",
                        "description": "Exact MCP tool name returned by search.",
                    },
                    "arguments": {
                        "type": "object",
                        "description": "JSON object forwarded to the MCP tool.",
                        "additionalProperties": True,
                    },
                },
                "required": ["tool_name", "arguments"],
                "additionalProperties": False,
            },
            effects=ToolEffects.external_side_effect(),
            approval=ToolApprovalRequirement.DERIVED,
            descriptor_type=LazyMcpCallToolDescriptor,
        ),
    )

build_mcp_runner(runner, external)

Build a runner whose catalog carries lazy MCP search/call tools.

The caller supplies an already-open native runner so any request-scoped model resolver or durable-port assembly has already happened. The agent Pod metadata is a shared singleton, so its catalog is augmented on a copy.copy. The actual external MCP descriptors stay hidden behind lazy meta-tools so a large MCP server does not flood the model request with every tool schema.

Source code in plugins/spakky-mcp/src/spakky/plugins/mcp/descriptor.py 
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
def build_mcp_runner(
    runner: AgentRunner,
    external: Sequence[AgentToolDescriptor],
) -> AgentRunner:
    """Build a runner whose catalog carries lazy MCP search/call tools.

    The caller supplies an already-open native runner so any request-scoped model
    resolver or durable-port assembly has already happened. The agent Pod
    metadata is a shared singleton, so its catalog is augmented on a ``copy.copy``.
    The actual external MCP descriptors stay hidden behind lazy meta-tools so a
    large MCP server does not flood the model request with every tool schema.
    """
    merged = merge_external_catalog(
        runner.agent.tool_catalog,
        build_lazy_mcp_descriptors(external),
    )
    augmented_agent = copy.copy(runner.agent)
    augmented_agent.tool_catalog = merged
    return dataclasses.replace(runner, agent=augmented_agent)

normalize_call_result(result)

Map an MCP tool result into a JSON value for evidence and the model.

Source code in plugins/spakky-mcp/src/spakky/plugins/mcp/descriptor.py 
286
287
288
289
290
291
292
293
294
295
def normalize_call_result(result: CallToolResult) -> JsonValue:
    """Map an MCP tool result into a JSON value for evidence and the model."""
    if result.isError:
        raise McpToolInvocationError("MCP tool reported an error result")
    if result.structuredContent is not None:
        return cast(JsonValue, result.structuredContent)
    texts = [block.text for block in result.content if isinstance(block, TextContent)]
    if not texts:
        raise McpResponseError("MCP tool result carries no readable content")
    return {"content": texts}

Plugin

Plugin initialization for the MCP external server adapter.

initialize(app)

Register MCP configuration and the external-tool runner factory.

Source code in plugins/spakky-mcp/src/spakky/plugins/mcp/main.py 
15
16
17
18
19
20
21
22
23
def initialize(app: SpakkyApplication) -> None:
    """Register MCP configuration and the external-tool runner factory."""
    app.add(McpConfig)
    app.add(McpHttpClientProvider)
    app.add(McpRuntimeServerResolver)
    app.add(McpClient)
    app.container.bind_to_type(IMcpHttpClientProvider, McpHttpClientProvider)
    app.container.bind_to_type(IMcpRuntimeServerResolver, McpRuntimeServerResolver)
    app.container.bind_to_type(IAgentRunnerFactory, McpClient)

에러

Error classes for the spakky-mcp external server adapter.

AbstractMcpError

Bases: AbstractSpakkyFrameworkError, ABC

Base class for MCP adapter errors.

McpServerConfigurationError

Bases: AbstractMcpError

Raised when an external MCP server declaration is invalid.

McpTransportError

Bases: AbstractMcpError

Raised when an external MCP server connection cannot be established.

McpToolDiscoveryError

Bases: AbstractMcpError

Raised when tool discovery against an MCP server fails.

McpToolInvocationError

Bases: AbstractMcpError

Raised when an external MCP tool call fails or reports an error result.

McpResponseError

Bases: AbstractMcpError

Raised when an MCP tool result cannot be mapped to a JSON value.

McpCatalogMergeError

Bases: AbstractMcpError

Raised when an external MCP tool collides with an existing catalog tool.