Changes
diff --git a/CMakeLists.txt b/CMakeLists.txt
index ea8fc1a..45827b1 100644
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -10,6 +10,7 @@ set(CMAKE_CXX_EXTENSIONS OFF)
option(AGENT_SMITH_BUILD_SHARED "Build agent_smith as a shared library" OFF)
option(AGENT_SMITH_BUILD_DEMO "Build the Agent Smith demo" OFF)
option(AGENT_SMITH_BUILD_TESTS "Build Agent Smith tests" OFF)
+option(AGENT_SMITH_TESTS_NEED_PYTHON "Run MCP subprocess tests" OFF)
option(AGENT_SMITH_TEST_ASAN "Enable AddressSanitizer for project tests" OFF)
option(AGENT_SMITH_TEST_TSAN "Enable ThreadSanitizer for project tests" OFF)
@@ -45,7 +46,7 @@ include(FetchContent)
FetchContent_Declare(
libmw
GIT_REPOSITORY https://github.com/MetroWind/libmw.git
- GIT_TAG 7336d71628d6e67f1720c36e8b6d7ffc78281821
+ GIT_TAG 78d8b9835e75ebf6a4d776d4c1c5805132f5ab9d
)
FetchContent_Declare(
@@ -77,6 +78,8 @@ add_library(agent_smith ${AGENT_SMITH_LIBRARY_TYPE}
src/llm_client.cpp
src/message.cpp
src/memory.cpp
+ src/mcp_client.cpp
+ src/transport_stdio.cpp
src/runtime.cpp
src/sse_parser.cpp
src/tool.cpp
diff --git a/demo/CMakeLists.txt b/demo/CMakeLists.txt
index 0e7eca6..c388099 100644
--- a/demo/CMakeLists.txt
+++ b/demo/CMakeLists.txt
@@ -1,3 +1,6 @@
add_executable(agent_smith_demo main.cpp)
+target_compile_definitions(agent_smith_demo PRIVATE
+ AGENT_SMITH_DEMO_MCP_SERVER="${CMAKE_CURRENT_SOURCE_DIR}/mcp_server_example.py"
+)
target_link_libraries(agent_smith_demo PRIVATE agent_smith::agent_smith)
diff --git a/demo/README.md b/demo/README.md
new file mode 100644
index 0000000..dacb687
--- /dev/null
+++ b/demo/README.md
@@ -0,0 +1,33 @@
+# Demo CLI
+
+Build with `-DAGENT_SMITH_BUILD_DEMO=ON`, then run
+`build/demo/agent_smith_demo`. Configure the LLM with `LLAMA_SERVER_URL`,
+`LLAMA_MODEL`, and optionally `LLAMA_API_KEY`.
+
+The CLI keeps rolling conversation history and exposes the native
+`get_weather` tool. It also starts `mcp_server_example.py` using Python 3
+and registers these MCP tools:
+
+- `mcp_demo__add`: add two numbers.
+- `mcp_demo__read_resource`: read `example://help`.
+
+Try “Use the addition tool to add 23 and 19” or “Read example://help”.
+The MCP greeting prompt is available through `mcp::Client::getPrompt()`;
+it is not automatically inserted into the conversation.
+
+Use `--no-mcp` to run without Python or an MCP server. To connect another
+stdio server, use `--mcp PROGRAM [ARG...]`; arguments are passed directly,
+not interpreted by a shell. Its tools use the same `mcp_demo__` prefix.
+
+Commands: `/help`, `/clear`, `/stream on`, `/stream off`, `/quit`.
+
+The bundled server uses only Python's standard library. It is a small
+protocol example, not a general-purpose MCP server implementation.
+
+## MCP tests
+
+Enable `-DAGENT_SMITH_BUILD_TESTS=ON
+-DAGENT_SMITH_TESTS_NEED_PYTHON=ON` to include the subprocess tests, then
+build with `cmake --build build -j24` and run `ctest --test-dir build
+--output-on-failure`. The Python tests are opt-in so builds without a
+Python interpreter retain the existing C++ test suite.
diff --git a/demo/main.cpp b/demo/main.cpp
index 51d7d87..87b8195 100644
--- a/demo/main.cpp
+++ b/demo/main.cpp
@@ -20,7 +20,7 @@ std::string envOr(std::string_view name, std::string_view fallback)
void printUsage(std::ostream& output)
{
- output << "Usage: agent_smith_demo\n"
+ output << "Usage: agent_smith_demo [--no-mcp | --mcp PROGRAM [ARG...]]\n"
<< "\n"
<< "Environment variables:\n"
<< " LLAMA_SERVER_URL Provider origin "
@@ -28,7 +28,8 @@ void printUsage(std::ostream& output)
<< " LLAMA_MODEL Model name (default: local-model)\n"
<< " LLAMA_API_KEY Optional bearer token\n"
<< "\n"
- << "The demo provides a tool for current weather by city.\n"
+ << "Tools: weather by city and the bundled Python MCP server.\n"
+ << "MCP provides addition, example://help, and a greeting prompt.\n"
<< "\n"
<< "Commands: /help, /clear, /stream on, /stream off, /quit\n";
}
@@ -185,6 +186,35 @@ int main(int argc, char* argv[])
agent.hooks().on_tool_result = printToolResult;
agent.hooks().on_message = printUsageReport;
+ if(argc <= 1 || std::string_view(argv[1]) != "--no-mcp")
+ {
+ std::vector<std::string> command{
+ "python3", AGENT_SMITH_DEMO_MCP_SERVER};
+ if(argc > 1)
+ {
+ if(std::string_view(argv[1]) != "--mcp" || argc < 3)
+ {
+ printUsage(std::cerr);
+ return 2;
+ }
+ command.assign(argv + 2, argv + argc);
+ }
+ agent_smith::mcp::ServerConfig mcp_config{
+ .alias = "demo",
+ .transport = agent_smith::mcp::StdioServerConfig{
+ std::move(command)},
+ };
+ auto connected = agent_smith::syncWait(
+ agent.connectMcp(std::move(mcp_config)));
+ if(!connected)
+ {
+ std::cerr << "MCP connection failed: " << connected.error().msg()
+ << "\nUse --no-mcp to run without an MCP server.\n";
+ return 2;
+ }
+ std::cout << "[MCP demo connected]\n";
+ }
+
std::cout << "Agent Smith " << agent_smith::VERSION
<< " agent CLI. Enter /help for commands.\n";
bool stream = true;
diff --git a/demo/mcp_server_example.py b/demo/mcp_server_example.py
new file mode 100644
index 0000000..e2911db
--- /dev/null
+++ b/demo/mcp_server_example.py
@@ -0,0 +1,84 @@
+"""Small stdlib-only MCP stdio example: addition, a resource, and a prompt.
+
+This is an example server, not an MCP server framework. Stdout contains
+only newline-delimited JSON-RPC frames; diagnostics belong on stderr.
+"""
+
+import json
+import sys
+
+
+def dispatch(method, params):
+ if method == "initialize":
+ return {
+ "protocolVersion": "2025-06-18",
+ "capabilities": {"tools": {}, "resources": {}, "prompts": {}},
+ "serverInfo": {"name": "agent-smith-example", "version": "1.0"},
+ }
+ if method == "ping":
+ return {}
+ if method == "tools/list":
+ return {"tools": [{
+ "name": "add",
+ "description": "Add two numbers exactly.",
+ "inputSchema": {
+ "type": "object",
+ "properties": {"a": {"type": "number"},
+ "b": {"type": "number"}},
+ "required": ["a", "b"],
+ "additionalProperties": False,
+ },
+ }]}
+ if method == "tools/call":
+ args = params.get("arguments", {})
+ if params.get("name") != "add":
+ raise ValueError("Unknown tool")
+ a, b = args.get("a"), args.get("b")
+ if (isinstance(a, bool) or isinstance(b, bool)
+ or not isinstance(a, (int, float))
+ or not isinstance(b, (int, float))):
+ return {"isError": True,
+ "content": [{"type": "text",
+ "text": "a and b must be numbers"}]}
+ result = a + b
+ return {"content": [{"type": "text", "text": str(result)}],
+ "structuredContent": {"sum": result}}
+ if method == "resources/list":
+ return {"resources": [{"uri": "example://help", "name": "Help",
+ "mimeType": "text/plain"}]}
+ if method == "resources/read":
+ if params.get("uri") != "example://help":
+ raise ValueError("Unknown resource")
+ return {"contents": [{"uri": "example://help",
+ "text": "The add tool adds two numbers."}]}
+ if method == "prompts/list":
+ return {"prompts": [{"name": "greet", "description": "Greet a name",
+ "arguments": [{"name": "name",
+ "required": True}]}]}
+ if method == "prompts/get":
+ if params.get("name") != "greet":
+ raise ValueError("Unknown prompt")
+ name = params.get("arguments", {}).get("name", "friend")
+ return {"messages": [{"role": "user", "content": {
+ "type": "text", "text": f"Say hello to {name}."}}]}
+ raise LookupError("Method not found")
+
+
+def main():
+ for line in sys.stdin:
+ request = json.loads(line)
+ if "id" not in request or "method" not in request:
+ continue
+ reply = {"jsonrpc": "2.0", "id": request["id"]}
+ try:
+ reply["result"] = dispatch(request["method"],
+ request.get("params", {}))
+ except (ValueError, KeyError) as error:
+ reply["error"] = {"code": -32602, "message": str(error)}
+ except LookupError as error:
+ reply["error"] = {"code": -32601, "message": str(error)}
+ print(json.dumps(reply), flush=True)
+
+
+if __name__ == "__main__":
+ main()
diff --git a/include/agent_smith/agent.hpp b/include/agent_smith/agent.hpp
index 8a3cc38..6b2e8fd 100644
--- a/include/agent_smith/agent.hpp
+++ b/include/agent_smith/agent.hpp
@@ -9,6 +9,7 @@
#include "llm_client.hpp"
#include "memory.hpp"
+#include "mcp_client.hpp"
namespace agent_smith
{
@@ -56,6 +57,8 @@ struct AgentHooks
std::function<void(const StreamEvent&)> on_stream_event;
/// Observe a framework error before send returns it.
std::function<void(const mw::Error&)> on_error;
+ /// Observe MCP notifications and connection transitions.
+ mcp::EventHook on_mcp_event;
};
/// One user turn with optional sampling and streaming overrides.
@@ -81,6 +84,10 @@ public:
void setSystemPrompt(std::string prompt);
/// Register a native tool; duplicate or empty names return an error.
mw::E<void> addTool(std::unique_ptr<Tool> tool, ToolOptions options = {});
+ /// Bind prefixed MCP tools and resource access; retain the client.
+ mw::E<void> addMcpClient(std::shared_ptr<mcp::Client> client);
+ /// Connect and bind an MCP server, observing initialization events.
+ mw::Task<mw::E<void>> connectMcp(mcp::ServerConfig config);
/// Access synchronous observers for this agent.
AgentHooks& hooks();
/// Cancel this agent and its shared tree; tools observe linked tokens and
@@ -96,7 +103,8 @@ private:
LlmClient* client_;
std::shared_ptr<Memory> memory_;
AgentConfig config_;
- AgentHooks hooks_;
+ std::shared_ptr<AgentHooks> hooks_ = std::make_shared<AgentHooks>();
+ std::vector<std::shared_ptr<mcp::Client>> mcp_clients_;
ToolRegistry tools_;
SystemMessage system_;
CancellationSource cancel_;
diff --git a/include/agent_smith/agent_smith.hpp b/include/agent_smith/agent_smith.hpp
index 7f3a85f..a328994 100644
--- a/include/agent_smith/agent_smith.hpp
+++ b/include/agent_smith/agent_smith.hpp
@@ -1,4 +1,5 @@
#pragma once
+#include "mcp_client.hpp"
#include "agent.hpp"
#include "cancellation.hpp"
diff --git a/include/agent_smith/error.hpp b/include/agent_smith/error.hpp
index 0661cb3..8b38c36 100644
--- a/include/agent_smith/error.hpp
+++ b/include/agent_smith/error.hpp
@@ -6,10 +6,26 @@
#include <string_view>
#include <mw/error.hpp>
+#include <nlohmann/json.hpp>
namespace agent_smith
{
+/// MCP transport, protocol, or remote JSON-RPC failure.
+struct McpError
+{
+ /// Human-readable transport, protocol, or remote failure.
+ std::string msg;
+ /// Configured server alias, when a connection is available.
+ std::string server = {};
+ /// Remote JSON-RPC error code, absent for local errors.
+ std::optional<int> rpc_code = std::nullopt;
+ /// Original remote error details.
+ nlohmann::json data = nullptr;
+ /// Whether reconnecting can repair the underlying connection.
+ bool transport = false;
+};
+
/// The common result type for Agent Smith operations that can fail.
template <typename T> using Result = mw::E<T>;
diff --git a/include/agent_smith/mcp_client.hpp b/include/agent_smith/mcp_client.hpp
new file mode 100644
index 0000000..06a696a
--- /dev/null
+++ b/include/agent_smith/mcp_client.hpp
@@ -0,0 +1,244 @@
+#pragma once
+
+#include <mw/url.hpp>
+
+#include "message.hpp"
+#include "tool.hpp"
+
+namespace agent_smith::mcp
+{
+
+/// An unsolicited server notification.
+struct Notification
+{
+ /// JSON-RPC notification method.
+ std::string method;
+ /// Original notification parameters.
+ nlohmann::json params;
+};
+/// Observable connection transitions.
+enum class LifecycleEventKind
+{
+ /// Initialization and discovery succeeded.
+ CONNECTED,
+ /// Transport failed or explicit shutdown began.
+ DISCONNECTED,
+ /// A replacement connection is being initialized.
+ RECONNECTING,
+ /// Replacement connection could not be initialized.
+ RECONNECT_FAILED
+};
+/// A connection transition for one configured server.
+struct LifecycleEvent
+{
+ /// Transition being reported.
+ LifecycleEventKind kind;
+ /// Configured server alias.
+ std::string server_alias;
+ /// Human-readable transition context.
+ std::string message;
+};
+/// Server notification or connection transition.
+using Event = std::variant<Notification, LifecycleEvent>;
+/// Best-effort observer; must not block or reenter the client.
+using EventHook = std::function<void(const Event&)>;
+/// JSON-RPC request envelope.
+struct Request
+{
+ /// Client-generated request identifier.
+ std::int64_t id;
+ /// Method being invoked.
+ std::string method;
+ /// Method parameters.
+ nlohmann::json params;
+};
+/// Remote JSON-RPC failure.
+struct RpcError
+{
+ /// Server's JSON-RPC error code.
+ int code;
+ /// Server's error description.
+ std::string message;
+ /// Optional original error details.
+ nlohmann::json data;
+};
+/// JSON-RPC response envelope.
+struct Response
+{
+ /// Identifier of the request being answered.
+ std::int64_t id;
+ /// Result payload or remote error.
+ std::variant<nlohmann::json, RpcError> body;
+};
+/// Server identity and original metadata.
+struct ServerInfo
+{
+ /// Reported implementation name.
+ std::string name;
+ /// Reported implementation version.
+ std::string version;
+ /// Original serverInfo object.
+ nlohmann::json raw;
+};
+/// One advertised resource.
+struct ResourceInfo
+{
+ /// Identifier passed to resources/read.
+ std::string uri;
+ /// Human-readable resource name.
+ std::string name;
+ /// Optional server description.
+ std::optional<std::string> description;
+ /// Optional advertised media type.
+ std::optional<std::string> mime_type;
+ /// Original resource descriptor.
+ nlohmann::json raw;
+};
+/// One advertised prompt template.
+struct PromptInfo
+{
+ /// Name passed to prompts/get.
+ std::string name;
+ /// Optional server description.
+ std::optional<std::string> description;
+ /// Original argument descriptors.
+ nlohmann::json arguments = nlohmann::json::array();
+ /// Original prompt descriptor.
+ nlohmann::json raw;
+};
+/// Text-only rendered prompt and original response.
+struct PromptRendering
+{
+ /// Rendered user and assistant text messages.
+ std::vector<Message> messages;
+ /// Original prompts/get response.
+ nlohmann::json raw;
+};
+/// Supported transport selectors; HTTP arrives in milestone 15.
+enum class McpTransportKind
+{
+ /// A child process using newline-delimited JSON on stdin/stdout.
+ STDIO,
+ /// Reserved for the future HTTP transport.
+ STREAMABLE_HTTP
+};
+/// Executable and arguments, without shell interpretation.
+struct StdioServerConfig
+{
+ /// Executable followed by arguments; passed directly without a shell.
+ std::vector<std::string> argv;
+};
+/// Streamable HTTP endpoint configuration.
+struct HttpServerConfig
+{
+ /// Future HTTP endpoint; not used by stdio.
+ mw::URL endpoint;
+};
+/// Connection, discovery, filtering and timeout policy.
+struct ServerConfig
+{
+ /// Unique alias used in local tool names; letters, digits, '_' and '-'.
+ std::string alias;
+ /// Transport selector; must match transport's active variant.
+ McpTransportKind kind = McpTransportKind::STDIO;
+ /// Transport-specific configuration.
+ std::variant<StdioServerConfig, HttpServerConfig> transport;
+ /// Reserved HTTP authentication; not used by stdio.
+ std::optional<std::string> bearer_token;
+ /// Reserved HTTP headers; not used by stdio.
+ std::vector<std::pair<std::string, std::string>> extra_headers;
+ /// Default budget for initialization, discovery, resources and prompts.
+ std::chrono::milliseconds request_timeout{30000};
+ /// Allowed remote tool names; empty permits all non-denied names.
+ std::vector<std::string> allow_tools;
+ /// Excluded remote names; takes precedence over allow_tools.
+ std::vector<std::string> deny_tools;
+ /// Fail connection when tools/list is unavailable.
+ bool tools_required = true;
+ /// Reconnect once for read-only failure or a later operation after loss.
+ bool reconnect_on_transport_error = true;
+};
+/// Bidirectional framed transport. Runtime must outlive the transport.
+class Transport
+{
+public:
+ /// Release transport resources.
+ virtual ~Transport() = default;
+ /// Send one frame within an absolute deadline. A failed write may have
+ /// delivered a prefix and must not be automatically replayed.
+ virtual mw::Task<mw::E<void>>
+ send(nlohmann::json frame,
+ std::chrono::steady_clock::time_point deadline =
+ std::chrono::steady_clock::now() + std::chrono::milliseconds(100),
+ CancellationToken cancel = {}) = 0;
+ /// Receive one frame or a typed failure.
+ virtual mw::Task<mw::E<nlohmann::json>> recv() = 0;
+ /// Wake an idle receiver without closing descriptors during active I/O.
+ virtual void interrupt() noexcept = 0;
+ /// Stop I/O and reap the child.
+ virtual mw::Task<mw::E<void>> close() = 0;
+};
+/// Concurrent MCP requests over one stdio connection.
+/// Runtime must outlive this client and all in-flight calls.
+/// Discovery getters must not be used while a reconnect is in progress.
+class Client
+{
+public:
+ /// Initialize and discover a server before returning it.
+ static mw::Task<mw::E<std::unique_ptr<Client>>>
+ connect(Runtime& runtime, ServerConfig config, EventHook hook = {});
+ /// Close owned transport, including when shutdown was omitted.
+ ~Client();
+ /// Negotiated server metadata.
+ const ServerInfo& info() const;
+ /// Filtered tool schemas, retaining original remote names.
+ const std::vector<ToolSchema>& tools() const;
+ /// Advertised resources.
+ const std::vector<ResourceInfo>& resources() const;
+ /// Advertised prompt templates.
+ const std::vector<PromptInfo>& prompts() const;
+ /// Stable configured server alias.
+ const std::string& alias() const;
+ /// Invoke a tool; never automatically replay a written request.
+ mw::Task<mw::E<ToolResult>> callTool(std::string name, nlohmann::json args,
+ std::chrono::milliseconds timeout,
+ CancellationToken cancel = {});
+ /// Read text resources; binary contents are unsupported.
+ mw::Task<mw::E<std::string>> readResource(
+ std::string uri, CancellationToken cancel = {},
+ std::optional<std::chrono::milliseconds> timeout = std::nullopt);
+ /// Render a text-only prompt without applying it to an agent.
+ mw::Task<mw::E<PromptRendering>> getPrompt(std::string name,
+ nlohmann::json args);
+ /// Replace the notification/lifecycle observer.
+ void setEventHook(EventHook hook);
+ /// Close connection and wake pending operations.
+ mw::Task<mw::E<void>> shutdown();
+
+private:
+ struct State;
+ explicit Client(std::shared_ptr<State> state);
+ static mw::Task<mw::E<void>> receive(std::shared_ptr<State> state);
+ mw::Task<mw::E<nlohmann::json>> request(std::string method,
+ nlohmann::json params,
+ std::chrono::milliseconds timeout,
+ CancellationToken cancel = {},
+ bool allow_reconnect = true);
+ mw::Task<mw::E<void>> discover();
+ mw::Task<mw::E<void>> reconnect(std::shared_ptr<State> previous);
+ std::mutex state_mutex_;
+ bool reconnecting_ = false;
+ bool stopped_ = false;
+ std::string alias_;
+ std::chrono::milliseconds default_timeout_;
+ std::shared_ptr<State> state_;
+ ServerInfo info_;
+ std::vector<ToolSchema> tools_;
+ std::vector<ResourceInfo> resources_;
+ std::vector<PromptInfo> prompts_;
+};
+
+/// Adapt all remote tools and a resource reader; retain the client safely.
+std::vector<std::unique_ptr<Tool>> adaptTools(std::shared_ptr<Client> client);
+
+} // namespace agent_smith::mcp
diff --git a/src/agent.cpp b/src/agent.cpp
index 5386c64..a41de5a 100644
--- a/src/agent.cpp
+++ b/src/agent.cpp
@@ -35,7 +35,52 @@ mw::E<void> Agent::addTool(std::unique_ptr<Tool> tool, ToolOptions options)
return {};
}
-AgentHooks& Agent::hooks() { return hooks_; }
+mw::E<void> Agent::addMcpClient(std::shared_ptr<mcp::Client> client)
+{
+ if(!client)
+ {
+ return std::unexpected(mw::runtimeError("null MCP client"));
+ }
+ auto adapters = mcp::adaptTools(client);
+ std::vector<std::string> names;
+ for(const auto& tool : adapters)
+ {
+ const auto& name = tool->schema().name;
+ if(tools_.find(name) || std::ranges::find(names, name) != names.end())
+ {
+ return std::unexpected(mw::runtimeError("duplicate MCP tool: " + name));
+ }
+ names.push_back(name);
+ }
+ std::weak_ptr<AgentHooks> observers = hooks_;
+ client->setEventHook([observers](const mcp::Event& event)
+ {
+ if(auto hooks = observers.lock())
+ {
+ internal::invokeHook(hooks->on_mcp_event, event);
+ }
+ });
+ for(auto& tool : adapters) { tools_.add(std::move(tool)); }
+ mcp_clients_.push_back(std::move(client));
+ return {};
+}
+
+mw::Task<mw::E<void>> Agent::connectMcp(mcp::ServerConfig config)
+{
+ std::weak_ptr<AgentHooks> observers = hooks_;
+ auto connected = co_await mcp::Client::connect(*runtime_, std::move(config),
+ [observers](const mcp::Event& event)
+ {
+ if(auto hooks = observers.lock())
+ {
+ internal::invokeHook(hooks->on_mcp_event, event);
+ }
+ });
+ if(!connected) { co_return std::unexpected(connected.error()); }
+ co_return addMcpClient(std::shared_ptr<mcp::Client>(std::move(*connected)));
+}
+
+AgentHooks& Agent::hooks() { return *hooks_; }
void Agent::cancel()
{
cancel_.cancel();
@@ -54,7 +99,7 @@ mw::Task<mw::E<AssistantMessage>> Agent::send(AgentRequest request)
auto result = co_await run(std::move(request));
if(!result)
{
- internal::invokeHook(hooks_.on_error, result.error());
+ internal::invokeHook(hooks_->on_error, result.error());
}
co_return result;
}
@@ -115,7 +160,7 @@ mw::Task<mw::E<AssistantMessage>> Agent::run(AgentRequest request)
auto response = request.stream.value_or(config_.stream_default)
? co_await client_->completeStream(std::move(llm_request),
[this](const StreamEvent& event)
- { internal::invokeHook(hooks_.on_stream_event, event); })
+ { internal::invokeHook(hooks_->on_stream_event, event); })
: co_await client_->complete(std::move(llm_request));
if(!response)
{
@@ -129,7 +174,7 @@ mw::Task<mw::E<AssistantMessage>> Agent::run(AgentRequest request)
total_tokens += std::min(remaining, message.usage->total);
}
memory_->append(message);
- internal::invokeHook(hooks_.on_message, Message(message));
+ internal::invokeHook(hooks_->on_message, Message(message));
if(auto error = stopped())
{
co_return std::unexpected(*error);
@@ -140,7 +185,7 @@ mw::Task<mw::E<AssistantMessage>> Agent::run(AgentRequest request)
}
for(const auto& call : message.tool_calls)
{
- internal::invokeHook(hooks_.on_tool_call, call);
+ internal::invokeHook(hooks_->on_tool_call, call);
}
if(auto error = stopped())
{
@@ -156,7 +201,7 @@ mw::Task<mw::E<AssistantMessage>> Agent::run(AgentRequest request)
for(const auto& result : *results)
{
memory_->append(result);
- internal::invokeHook(hooks_.on_tool_result, result);
+ internal::invokeHook(hooks_->on_tool_result, result);
}
if(auto error = stopped())
{
@@ -167,7 +212,7 @@ mw::Task<mw::E<AssistantMessage>> Agent::run(AgentRequest request)
AssistantMessage final;
final.content = std::move(*final_message);
memory_->append(final);
- internal::invokeHook(hooks_.on_message, Message(final));
+ internal::invokeHook(hooks_->on_message, Message(final));
co_return final;
}
}
diff --git a/src/detail/transport_stdio.hpp b/src/detail/transport_stdio.hpp
new file mode 100644
index 0000000..1409b65
--- /dev/null
+++ b/src/detail/transport_stdio.hpp
@@ -0,0 +1,10 @@
+#pragma once
+
+#include "agent_smith/mcp_client.hpp"
+
+namespace agent_smith::mcp::internal
+{
+// Create a bounded, newline-delimited stdio transport.
+mw::E<std::unique_ptr<Transport>> openStdio(Runtime& runtime,
+ const StdioServerConfig& config);
+} // namespace agent_smith::mcp::internal
diff --git a/src/mcp_client.cpp b/src/mcp_client.cpp
new file mode 100644
index 0000000..4a242b5
--- /dev/null
+++ b/src/mcp_client.cpp
@@ -0,0 +1,891 @@
+#include "agent_smith/mcp_client.hpp"
+
+#include <algorithm>
+#include <map>
+#include <set>
+
+#include "agent_smith/detail/background_tasks.hpp"
+#include "agent_smith/detail/hook.hpp"
+#include "agent_smith/version.hpp"
+#include "detail/transport_stdio.hpp"
+
+namespace agent_smith::mcp
+{
+namespace
+{
+using Json = nlohmann::json;
+using Clock = std::chrono::steady_clock;
+using namespace std::chrono_literals;
+
+mw::Error protocolError(std::string message)
+{
+ return McpError{std::move(message)};
+}
+
+bool transportError(const mw::Error& error)
+{
+ const auto* mcp = error.as<McpError>();
+ return mcp && mcp->transport;
+}
+
+std::optional<std::string> optionalString(const Json& value, const char* key)
+{
+ if(!value.contains(key))
+ {
+ return std::nullopt;
+ }
+ return value.at(key).get<std::string>();
+}
+
+mw::E<std::string> textContent(const Json& blocks)
+{
+ if(!blocks.is_array())
+ {
+ return std::unexpected(protocolError("content must be an array"));
+ }
+ std::string text;
+ for(const auto& block : blocks)
+ {
+ if(!block.is_object() || block.value("type", "") != "text" ||
+ !block.contains("text") || !block["text"].is_string())
+ {
+ return std::unexpected(
+ protocolError("only text MCP content is supported"));
+ }
+ if(!text.empty())
+ {
+ text += '\n';
+ }
+ text += block["text"].get<std::string>();
+ }
+ return text;
+}
+} // namespace
+
+struct Client::State
+{
+ Runtime* runtime;
+ ServerConfig config;
+ std::unique_ptr<Transport> transport;
+ std::mutex mutex;
+ EventHook hook;
+ std::atomic<bool> stop{false};
+ std::optional<mw::Error> failure;
+ std::int64_t next_id = 1;
+ struct Pending
+ {
+ std::optional<mw::E<Json>> response;
+ };
+ std::map<std::int64_t, std::shared_ptr<Pending>> pending;
+
+ void emit(Event event)
+ {
+ EventHook observer;
+ {
+ std::lock_guard lock(mutex);
+ observer = hook;
+ }
+ agent_smith::internal::invokeHook(observer, event);
+ }
+
+ void fail(mw::Error error)
+ {
+ if(const auto* original = error.as<McpError>())
+ {
+ auto contextual = *original;
+ contextual.server = config.alias;
+ error = std::move(contextual);
+ }
+ bool first;
+ {
+ std::lock_guard lock(mutex);
+ first = !failure.has_value();
+ if(first)
+ {
+ failure = error;
+ }
+ }
+ if(first)
+ {
+ transport->interrupt();
+ emit(LifecycleEvent{LifecycleEventKind::DISCONNECTED, config.alias,
+ error.msg()});
+ }
+ }
+};
+
+Client::Client(std::shared_ptr<State> state)
+ : alias_(state->config.alias),
+ default_timeout_(state->config.request_timeout), state_(std::move(state))
+{
+}
+
+Client::~Client()
+{
+ if(state_)
+ {
+ state_->stop = true;
+ state_->transport->interrupt();
+ }
+}
+
+const ServerInfo& Client::info() const
+{
+ return info_;
+}
+const std::vector<ToolSchema>& Client::tools() const
+{
+ return tools_;
+}
+const std::vector<ResourceInfo>& Client::resources() const
+{
+ return resources_;
+}
+const std::vector<PromptInfo>& Client::prompts() const
+{
+ return prompts_;
+}
+const std::string& Client::alias() const
+{
+ return alias_;
+}
+
+void Client::setEventHook(EventHook hook)
+{
+ std::lock_guard client_lock(state_mutex_);
+ std::lock_guard lock(state_->mutex);
+ state_->hook = std::move(hook);
+}
+
+mw::Task<mw::E<std::unique_ptr<Client>>>
+Client::connect(Runtime& runtime, ServerConfig config, EventHook hook)
+{
+ if(config.alias.empty() ||
+ !std::all_of(config.alias.begin(), config.alias.end(),
+ [](char c)
+ {
+ return (c >= 'a' && c <= 'z') ||
+ (c >= 'A' && c <= 'Z') ||
+ (c >= '0' && c <= '9') || c == '_' || c == '-';
+ }) ||
+ config.request_timeout <= 0ms)
+ {
+ co_return std::unexpected(protocolError(
+ "MCP needs a nonempty function-safe alias and positive timeout"));
+ }
+ if(config.kind != McpTransportKind::STDIO ||
+ !std::holds_alternative<StdioServerConfig>(config.transport))
+ {
+ co_return std::unexpected(
+ protocolError("only MCP stdio transport is implemented"));
+ }
+ auto transport = internal::openStdio(
+ runtime, std::get<StdioServerConfig>(config.transport));
+ if(!transport)
+ {
+ co_return std::unexpected(transport.error());
+ }
+ auto state = std::make_shared<State>();
+ state->runtime = &runtime;
+ state->config = std::move(config);
+ state->transport = std::move(*transport);
+ state->hook = std::move(hook);
+ auto client = std::unique_ptr<Client>(new Client(state));
+ agent_smith::internal::BackgroundTasks::start(runtime, receive(state));
+ auto discovered = co_await client->discover();
+ if(!discovered)
+ {
+ co_await client->shutdown();
+ co_return std::unexpected(discovered.error());
+ }
+ state->emit(LifecycleEvent{LifecycleEventKind::CONNECTED,
+ state->config.alias, "MCP initialized"});
+ co_return client;
+}
+
+mw::Task<mw::E<void>> Client::receive(std::shared_ptr<State> state)
+{
+ while(!state->stop)
+ {
+ auto inbound = co_await state->transport->recv();
+ if(!inbound)
+ {
+ if(!state->stop)
+ {
+ state->fail(inbound.error());
+ }
+ break;
+ }
+ const Json& frame = *inbound;
+ try
+ {
+ if(!frame.is_object() || frame.value("jsonrpc", "") != "2.0")
+ {
+ state->fail(protocolError("invalid JSON-RPC envelope"));
+ break;
+ }
+ if(frame.contains("method"))
+ {
+ const auto method = frame.at("method").get<std::string>();
+ if(frame.contains("result") || frame.contains("error"))
+ {
+ state->fail(protocolError("invalid JSON-RPC request"));
+ break;
+ }
+ if(frame.contains("params") && !frame["params"].is_object() &&
+ !frame["params"].is_array())
+ {
+ state->fail(protocolError("invalid JSON-RPC params"));
+ break;
+ }
+ if(!frame.contains("id"))
+ {
+ state->emit(Notification{
+ method, frame.value("params", Json::object())});
+ }
+ else
+ {
+ if(!frame["id"].is_string() &&
+ !frame["id"].is_number_integer())
+ {
+ state->fail(protocolError("invalid server request id"));
+ break;
+ }
+ Json reply{{"jsonrpc", "2.0"}, {"id", frame["id"]}};
+ if(method == "ping")
+ {
+ reply["result"] = Json::object();
+ }
+ else
+ {
+ reply["error"] = {
+ {"code", -32601},
+ {"message", "client method not supported"}};
+ }
+ auto sent =
+ co_await state->transport->send(std::move(reply));
+ if(!sent)
+ {
+ state->fail(sent.error());
+ break;
+ }
+ }
+ continue;
+ }
+ if(!frame.contains("id") || !frame["id"].is_number_integer() ||
+ frame.contains("result") == frame.contains("error"))
+ {
+ state->fail(protocolError("invalid JSON-RPC response"));
+ break;
+ }
+ const auto id = frame.at("id").get<std::int64_t>();
+ mw::E<Json> response = Json{};
+ if(frame.contains("error"))
+ {
+ const auto& error = frame.at("error");
+ response = std::unexpected(
+ McpError{error.at("message").get<std::string>(),
+ state->config.alias, error.at("code").get<int>(),
+ error.value("data", Json(nullptr))});
+ }
+ else
+ {
+ response = frame.at("result");
+ }
+ std::lock_guard lock(state->mutex);
+ const auto found = state->pending.find(id);
+ // Replies to timed-out requests are intentionally discarded.
+ if(found != state->pending.end() && !found->second->response)
+ {
+ found->second->response = std::move(response);
+ }
+ }
+ catch(const Json::exception&)
+ {
+ state->fail(protocolError("malformed JSON-RPC fields"));
+ break;
+ }
+ }
+ co_return co_await state->transport->close();
+}
+
+mw::Task<mw::E<void>> Client::reconnect(std::shared_ptr<State> previous)
+{
+ while(true)
+ {
+ bool acquired = false;
+ {
+ std::lock_guard lock(state_mutex_);
+ if(stopped_)
+ {
+ co_return std::unexpected(
+ protocolError("MCP client shut down"));
+ }
+ if(state_ != previous)
+ {
+ co_return mw::E<void>{};
+ }
+ if(!reconnecting_)
+ {
+ reconnecting_ = true;
+ acquired = true;
+ }
+ }
+ if(acquired)
+ {
+ break;
+ }
+ co_await previous->runtime->sleepFor(2ms);
+ }
+ EventHook hook;
+ {
+ std::lock_guard lock(previous->mutex);
+ hook = previous->hook;
+ }
+ previous->emit(LifecycleEvent{LifecycleEventKind::RECONNECTING, alias_,
+ "reconnecting MCP transport"});
+ auto fresh = co_await connect(*previous->runtime, previous->config, hook);
+ {
+ std::lock_guard lock(state_mutex_);
+ reconnecting_ = false;
+ if(fresh && !stopped_)
+ {
+ state_ = std::move((*fresh)->state_);
+ // Existing adapters retain their advertised schemas; callers may
+ // inspect refreshed discovery when no requests are in flight.
+ info_ = std::move((*fresh)->info_);
+ tools_ = std::move((*fresh)->tools_);
+ resources_ = std::move((*fresh)->resources_);
+ prompts_ = std::move((*fresh)->prompts_);
+ }
+ else if(fresh)
+ {
+ co_return std::unexpected(protocolError("MCP client shut down"));
+ }
+ }
+ if(!fresh)
+ {
+ previous->emit(LifecycleEvent{LifecycleEventKind::RECONNECT_FAILED,
+ alias_, fresh.error().msg()});
+ co_return std::unexpected(fresh.error());
+ }
+ co_return mw::E<void>{};
+}
+
+mw::Task<mw::E<Json>> Client::request(std::string method, Json params,
+ std::chrono::milliseconds timeout,
+ CancellationToken cancel,
+ bool allow_reconnect)
+{
+ const auto deadline = Clock::now() + timeout;
+ for(int attempt = 0; attempt < 2; ++attempt)
+ {
+ std::shared_ptr<State> state;
+ {
+ std::lock_guard lock(state_mutex_);
+ if(stopped_)
+ {
+ co_return std::unexpected(
+ protocolError("MCP client shut down"));
+ }
+ state = state_;
+ }
+ std::optional<mw::Error> failure;
+ {
+ std::lock_guard lock(state->mutex);
+ failure = state->failure;
+ }
+ if(failure)
+ {
+ if(attempt == 0 && allow_reconnect &&
+ state->config.reconnect_on_transport_error &&
+ transportError(*failure))
+ {
+ auto result = co_await reconnect(state);
+ if(!result)
+ {
+ co_return std::unexpected(result.error());
+ }
+ continue;
+ }
+ co_return std::unexpected(*failure);
+ }
+ if(cancel.cancelled())
+ {
+ co_return std::unexpected(
+ CancelledError{"MCP request cancelled", method});
+ }
+ if(Clock::now() >= deadline)
+ {
+ co_return std::unexpected(
+ TimeoutError{"MCP request timed out", method, timeout});
+ }
+ auto pending = std::make_shared<State::Pending>();
+ std::int64_t id;
+ {
+ std::lock_guard lock(state->mutex);
+ id = state->next_id++;
+ state->pending.emplace(id, pending);
+ }
+ Json frame{{"jsonrpc", "2.0"},
+ {"id", id},
+ {"method", method},
+ {"params", params}};
+ auto sent =
+ co_await state->transport->send(std::move(frame), deadline, cancel);
+ if(!sent)
+ {
+ state->fail(sent.error());
+ }
+ mw::E<Json> result = std::unexpected(protocolError("no MCP response"));
+ bool abandoned = false;
+ if(!sent)
+ {
+ if(cancel.cancelled())
+ {
+ result = std::unexpected(CancelledError{
+ "MCP request cancelled during write", method});
+ }
+ else if(Clock::now() >= deadline)
+ {
+ result = std::unexpected(
+ TimeoutError{"MCP write timed out", method, timeout});
+ }
+ else
+ {
+ result = std::unexpected(sent.error());
+ }
+ }
+ while(sent)
+ {
+ bool ready = false;
+ {
+ std::lock_guard lock(state->mutex);
+ if(pending->response)
+ {
+ result = std::move(*pending->response);
+ ready = true;
+ }
+ else if(state->failure)
+ {
+ result = std::unexpected(*state->failure);
+ ready = true;
+ }
+ }
+ if(ready)
+ {
+ break;
+ }
+ if(state->stop || cancel.cancelled())
+ {
+ result = std::unexpected(
+ CancelledError{"MCP request cancelled", method});
+ abandoned = true;
+ break;
+ }
+ if(Clock::now() >= deadline)
+ {
+ result = std::unexpected(
+ TimeoutError{"MCP request timed out", method, timeout});
+ abandoned = true;
+ break;
+ }
+ co_await state->runtime->sleepFor(2ms);
+ }
+ {
+ std::lock_guard lock(state->mutex);
+ state->pending.erase(id);
+ }
+ if(abandoned && !state->stop && method != "initialize")
+ {
+ Json notification{
+ {"jsonrpc", "2.0"},
+ {"method", "notifications/cancelled"},
+ {"params",
+ {{"requestId", id}, {"reason", "client stopped waiting"}}}};
+ co_await state->transport->send(std::move(notification),
+ Clock::now() + 2ms);
+ }
+ if(!result && attempt == 0 && allow_reconnect &&
+ state->config.reconnect_on_transport_error &&
+ transportError(result.error()) && method != "tools/call")
+ {
+ auto connected = co_await reconnect(state);
+ if(!connected)
+ {
+ co_return std::unexpected(connected.error());
+ }
+ continue;
+ }
+ co_return result;
+ }
+ co_return std::unexpected(protocolError("MCP retry exhausted"));
+}
+
+mw::Task<mw::E<void>> Client::discover()
+{
+ const auto timeout = state_->config.request_timeout;
+ Json init{{"protocolVersion", "2025-06-18"},
+ {"capabilities", Json::object()},
+ {"clientInfo", {{"name", "agent_smith"}, {"version", VERSION}}}};
+ auto initialized =
+ co_await request("initialize", std::move(init), timeout, {}, false);
+ if(!initialized)
+ {
+ co_return std::unexpected(initialized.error());
+ }
+ try
+ {
+ if(initialized->at("protocolVersion") != "2025-06-18" ||
+ !initialized->at("capabilities").is_object())
+ {
+ co_return std::unexpected(
+ protocolError("unsupported MCP negotiation"));
+ }
+ const auto& info = initialized->at("serverInfo");
+ info_ = {info.at("name").get<std::string>(),
+ info.at("version").get<std::string>(), info};
+ }
+ catch(const Json::exception&)
+ {
+ co_return std::unexpected(protocolError("invalid initialize result"));
+ }
+ Json notification{{"jsonrpc", "2.0"},
+ {"method", "notifications/initialized"}};
+ auto sent = co_await state_->transport->send(std::move(notification));
+ if(!sent)
+ {
+ co_return std::unexpected(sent.error());
+ }
+ for(const std::string category : {"tools", "resources", "prompts"})
+ {
+ std::set<std::string> cursors;
+ Json params = Json::object();
+ for(int page = 0; page < 1024; ++page)
+ {
+ auto listed = co_await request(category + "/list", params, timeout,
+ {}, false);
+ if(!listed)
+ {
+ const auto* error = listed.error().as<McpError>();
+ const bool optional =
+ category != "tools" || !state_->config.tools_required;
+ const bool unsupported = error && error->rpc_code == -32601;
+ if(optional && error && !error->transport &&
+ (unsupported || category == "tools") && page == 0)
+ {
+ spdlog::warn("MCP {} does not provide {}: {}",
+ alias_.c_str(), category.c_str(),
+ error->msg.c_str());
+ break;
+ }
+ co_return std::unexpected(listed.error());
+ }
+ try
+ {
+ const auto& entries = listed->at(category);
+ if(!entries.is_array())
+ {
+ co_return std::unexpected(
+ protocolError("invalid discovery list"));
+ }
+ for(const auto& entry : entries)
+ {
+ const auto name = entry.at("name").get<std::string>();
+ if(category == "tools")
+ {
+ const auto& schema = entry.at("inputSchema");
+ if(!schema.is_object() || name.empty())
+ {
+ co_return std::unexpected(
+ protocolError("invalid tool schema"));
+ }
+ const auto& allow = state_->config.allow_tools;
+ const auto& deny = state_->config.deny_tools;
+ if((allow.empty() ||
+ std::ranges::find(allow, name) != allow.end()) &&
+ std::ranges::find(deny, name) == deny.end())
+ {
+ if(std::ranges::any_of(
+ tools_, [&name](const auto& tool)
+ { return tool.name == name; }))
+ {
+ co_return std::unexpected(
+ protocolError("duplicate MCP tool"));
+ }
+ tools_.push_back(
+ {name, entry.value("description", ""), schema});
+ }
+ }
+ else if(category == "resources")
+ {
+ resources_.push_back(
+ {entry.at("uri").get<std::string>(), name,
+ optionalString(entry, "description"),
+ optionalString(entry, "mimeType"), entry});
+ }
+ else
+ {
+ auto arguments =
+ entry.value("arguments", Json::array());
+ if(!arguments.is_array())
+ {
+ co_return std::unexpected(
+ protocolError("invalid prompt arguments"));
+ }
+ prompts_.push_back(
+ {name, optionalString(entry, "description"),
+ arguments, entry});
+ }
+ }
+ if(!listed->contains("nextCursor"))
+ {
+ break;
+ }
+ auto cursor = listed->at("nextCursor").get<std::string>();
+ if(!cursors.insert(cursor).second || page == 1023)
+ {
+ co_return std::unexpected(
+ protocolError("discovery pagination cycle or limit"));
+ }
+ params = {{"cursor", cursor}};
+ }
+ catch(const Json::exception&)
+ {
+ co_return std::unexpected(
+ protocolError("malformed MCP discovery result"));
+ }
+ }
+ }
+ co_return mw::E<void>{};
+}
+
+mw::Task<mw::E<ToolResult>> Client::callTool(std::string name, Json args,
+ std::chrono::milliseconds timeout,
+ CancellationToken cancel)
+{
+ std::shared_ptr<State> state;
+ {
+ std::lock_guard lock(state_mutex_);
+ state = state_;
+ }
+ const auto& allow = state->config.allow_tools;
+ const auto& deny = state->config.deny_tools;
+ if((!allow.empty() && std::ranges::find(allow, name) == allow.end()) ||
+ std::ranges::find(deny, name) != deny.end())
+ {
+ co_return std::unexpected(protocolError("MCP tool excluded by filter"));
+ }
+ if(!args.is_object())
+ {
+ co_return std::unexpected(
+ protocolError("MCP tool arguments must be an object"));
+ }
+ auto result = co_await request(
+ "tools/call",
+ Json{{"name", std::move(name)}, {"arguments", std::move(args)}},
+ timeout, std::move(cancel));
+ if(!result)
+ {
+ co_return std::unexpected(result.error());
+ }
+ try
+ {
+ auto text = textContent(result->at("content"));
+ if(!text)
+ {
+ co_return std::unexpected(text.error());
+ }
+ co_return ToolResult{std::move(*text), result->value("isError", false),
+ result->value("structuredContent", Json(nullptr))};
+ }
+ catch(const Json::exception&)
+ {
+ co_return std::unexpected(protocolError("invalid MCP tool result"));
+ }
+}
+
+mw::Task<mw::E<std::string>>
+Client::readResource(std::string uri, CancellationToken cancel,
+ std::optional<std::chrono::milliseconds> timeout)
+{
+ auto result =
+ co_await request("resources/read", Json{{"uri", std::move(uri)}},
+ timeout.value_or(default_timeout_), cancel);
+ if(!result)
+ {
+ co_return std::unexpected(result.error());
+ }
+ try
+ {
+ if(!result->at("contents").is_array())
+ {
+ co_return std::unexpected(
+ protocolError("invalid resource contents"));
+ }
+ std::string text;
+ for(const auto& content : result->at("contents"))
+ {
+ if(content.contains("blob") || !content.at("text").is_string())
+ {
+ co_return std::unexpected(
+ protocolError("binary MCP resources unsupported"));
+ }
+ if(!text.empty())
+ {
+ text += '\n';
+ }
+ text += content.at("text").get<std::string>();
+ }
+ co_return text;
+ }
+ catch(const Json::exception&)
+ {
+ co_return std::unexpected(protocolError("invalid MCP resource result"));
+ }
+}
+
+mw::Task<mw::E<PromptRendering>> Client::getPrompt(std::string name, Json args)
+{
+ if(!args.is_object())
+ {
+ co_return std::unexpected(
+ protocolError("prompt arguments must be an object"));
+ }
+ auto result = co_await request(
+ "prompts/get",
+ Json{{"name", std::move(name)}, {"arguments", std::move(args)}},
+ default_timeout_);
+ if(!result)
+ {
+ co_return std::unexpected(result.error());
+ }
+ try
+ {
+ PromptRendering rendering{{}, *result};
+ if(!result->at("messages").is_array())
+ {
+ co_return std::unexpected(protocolError("invalid prompt messages"));
+ }
+ for(const auto& message : result->at("messages"))
+ {
+ auto text = textContent(Json::array({message.at("content")}));
+ if(!text)
+ {
+ co_return std::unexpected(text.error());
+ }
+ const auto role = message.at("role").get<std::string>();
+ if(role == "user")
+ {
+ rendering.messages.push_back(UserMessage{*text});
+ }
+ else if(role == "assistant")
+ {
+ AssistantMessage assistant;
+ assistant.content = *text;
+ rendering.messages.push_back(std::move(assistant));
+ }
+ else
+ {
+ co_return std::unexpected(
+ protocolError("invalid MCP prompt role"));
+ }
+ }
+ co_return rendering;
+ }
+ catch(const Json::exception&)
+ {
+ co_return std::unexpected(protocolError("invalid MCP prompt result"));
+ }
+}
+
+mw::Task<mw::E<void>> Client::shutdown()
+{
+ std::shared_ptr<State> state;
+ {
+ std::lock_guard lock(state_mutex_);
+ stopped_ = true;
+ state = state_;
+ }
+ state->stop = true;
+ state->transport->interrupt();
+ auto result = co_await state->transport->close();
+ state->fail(
+ McpError{"MCP client shut down", alias_, std::nullopt, nullptr, true});
+ co_return result;
+}
+
+namespace
+{
+class RemoteTool final : public Tool
+{
+public:
+ RemoteTool(std::shared_ptr<Client> client, ToolSchema schema,
+ bool resource = false)
+ : client_(std::move(client)), schema_(std::move(schema)),
+ remote_name_(schema_.name), resource_(resource)
+ {
+ schema_.name = "mcp_" + client_->alias() + "__" + schema_.name;
+ }
+ const ToolSchema& schema() const override
+ {
+ return schema_;
+ }
+ mw::Task<mw::E<ToolResult>> invoke(Json args, ToolContext context) override
+ {
+ if(resource_)
+ {
+ if(!args.is_object() || !args.contains("uri") ||
+ !args["uri"].is_string())
+ {
+ co_return std::unexpected(
+ protocolError("resource tool requires uri"));
+ }
+ if(context.cancel.cancelled())
+ {
+ co_return std::unexpected(CancelledError{
+ "resource read cancelled", "resources/read"});
+ }
+ auto result = co_await client_->readResource(
+ args["uri"].get<std::string>(), context.cancel, context.budget);
+ if(!result)
+ {
+ co_return std::unexpected(result.error());
+ }
+ co_return ToolResult{.content = std::move(*result)};
+ }
+ co_return co_await client_->callTool(remote_name_, std::move(args),
+ context.budget, context.cancel);
+ }
+
+private:
+ std::shared_ptr<Client> client_;
+ ToolSchema schema_;
+ std::string remote_name_;
+ bool resource_;
+};
+} // namespace
+
+std::vector<std::unique_ptr<Tool>> adaptTools(std::shared_ptr<Client> client)
+{
+ std::vector<std::unique_ptr<Tool>> tools;
+ if(!client)
+ {
+ return tools;
+ }
+ for(const auto& schema : client->tools())
+ {
+ tools.push_back(std::make_unique<RemoteTool>(client, schema));
+ }
+ tools.push_back(std::make_unique<RemoteTool>(
+ client,
+ ToolSchema{"read_resource",
+ "Read a text resource from the MCP server.",
+ {{"type", "object"},
+ {"properties", {{"uri", {{"type", "string"}}}}},
+ {"required", {"uri"}},
+ {"additionalProperties", false}}},
+ true));
+ return tools;
+}
+} // namespace agent_smith::mcp
diff --git a/src/transport_stdio.cpp b/src/transport_stdio.cpp
new file mode 100644
index 0000000..3ce865f
--- /dev/null
+++ b/src/transport_stdio.cpp
@@ -0,0 +1,201 @@
+#include "detail/transport_stdio.hpp"
+
+#include <mw/exec.hpp>
+
+namespace agent_smith::mcp::internal
+{
+namespace
+{
+mw::Error failure(std::string message)
+{
+ return McpError{std::move(message), {}, std::nullopt, nullptr, true};
+}
+
+class StdioTransport final : public Transport
+{
+public:
+ StdioTransport(Runtime& runtime, mw::Pipe input, mw::Pipe output,
+ mw::Process process)
+ : runtime_(runtime), input_(std::move(input)),
+ output_(std::move(output)), process_(std::move(process))
+ {
+ }
+
+ mw::Task<mw::E<void>> send(nlohmann::json frame,
+ std::chrono::steady_clock::time_point deadline,
+ CancellationToken cancel) override
+ {
+ std::string bytes = frame.dump() + "\n";
+ std::function<mw::E<void>()> work =
+ [this, bytes = std::move(bytes), deadline, cancel]
+ {
+ std::unique_lock lock(write_mutex_, std::defer_lock);
+ while(!lock.try_lock_for(std::chrono::milliseconds(2)))
+ {
+ if(cancel.cancelled() || interrupted_ ||
+ std::chrono::steady_clock::now() >= deadline)
+ {
+ return mw::E<void>(std::unexpected(failure(
+ "stdio write stopped before lock acquisition")));
+ }
+ }
+ if(closed_)
+ {
+ return mw::E<void>(std::unexpected(failure("stdio closed")));
+ }
+ std::size_t offset = 0;
+ while(offset < bytes.size())
+ {
+ const auto now = std::chrono::steady_clock::now();
+ if(cancel.cancelled() || interrupted_ || now >= deadline)
+ {
+ return mw::E<void>(std::unexpected(failure(
+ "stdio write stopped (request may be partial)")));
+ }
+ auto result = input_.writeUntil(
+ std::string_view(bytes).substr(offset),
+ std::min(deadline, now + std::chrono::milliseconds(2)));
+ if(!result)
+ {
+ return mw::E<void>(std::unexpected(failure(
+ "stdio write failed (request may be partial)")));
+ }
+ offset += result->bytes_written;
+ }
+ return mw::E<void>{};
+ };
+ co_return co_await runtime_.runOnWorker(std::move(work));
+ }
+
+ mw::Task<mw::E<nlohmann::json>> recv() override
+ {
+ while(true)
+ {
+ if(interrupted_)
+ {
+ co_return std::unexpected(failure("stdio interrupted"));
+ }
+ const auto newline = buffer_.find('\n');
+ if(newline != std::string::npos)
+ {
+ auto frame = nlohmann::json::parse(buffer_.substr(0, newline),
+ nullptr, false);
+ buffer_.erase(0, newline + 1);
+ if(frame.is_discarded())
+ {
+ co_return std::unexpected(
+ McpError{"invalid stdio JSON frame"});
+ }
+ co_return frame;
+ }
+ std::function<mw::E<mw::PipeReadResult>()> work = [this]
+ {
+ std::lock_guard lock(read_mutex_);
+ if(closed_)
+ {
+ return mw::E<mw::PipeReadResult>(
+ std::unexpected(failure("stdio closed")));
+ }
+ return output_.tryRead(8192);
+ };
+ auto bytes = co_await runtime_.runOnWorker(std::move(work));
+ if(!bytes)
+ {
+ co_return std::unexpected(failure(bytes.error().msg()));
+ }
+ if(bytes->state == mw::PipeState::EOF_REACHED)
+ {
+ co_return std::unexpected(failure("MCP server closed stdout"));
+ }
+ buffer_ += bytes->data;
+ if(buffer_.size() > 4 * 1024 * 1024)
+ {
+ co_return std::unexpected(McpError{"MCP frame too large"});
+ }
+ if(bytes->state == mw::PipeState::WOULD_BLOCK)
+ {
+ co_await runtime_.sleepFor(std::chrono::milliseconds(2));
+ }
+ }
+ }
+
+ mw::Task<mw::E<void>> close() override
+ {
+ interrupt();
+ std::function<mw::E<void>()> work = [this]
+ {
+ std::scoped_lock lock(read_mutex_, write_mutex_);
+ if(closed_)
+ {
+ return mw::E<void>{};
+ }
+ closed_ = true;
+ std::ignore = input_.closeInput();
+ std::ignore = output_.closeOutput();
+ auto exited = process_.waitFor(std::chrono::milliseconds(50));
+ if(!exited || exited->state != mw::ProcessWaitState::COMPLETE)
+ {
+ exited = process_.shutdown(std::chrono::milliseconds(50),
+ std::chrono::steady_clock::now() +
+ std::chrono::milliseconds(500));
+ }
+ if(!exited)
+ {
+ return mw::E<void>(
+ std::unexpected(failure(exited.error().msg())));
+ }
+ if(exited->state != mw::ProcessWaitState::COMPLETE)
+ {
+ return mw::E<void>(std::unexpected(
+ failure("MCP child shutdown deadline expired")));
+ }
+ return mw::E<void>{};
+ };
+ co_return co_await runtime_.runOnWorker(std::move(work));
+ }
+
+ void interrupt() noexcept override
+ {
+ interrupted_ = true;
+ }
+
+private:
+ Runtime& runtime_;
+ mw::Pipe input_;
+ mw::Pipe output_;
+ mw::Process process_;
+ std::mutex read_mutex_;
+ std::timed_mutex write_mutex_;
+ bool closed_ = false;
+ std::atomic<bool> interrupted_{false};
+ std::string buffer_;
+};
+} // namespace
+
+mw::E<std::unique_ptr<Transport>> openStdio(Runtime& runtime,
+ const StdioServerConfig& config)
+{
+ auto input = mw::Pipe::create();
+ auto output = mw::Pipe::create();
+ if(!input)
+ {
+ return std::unexpected(failure(input.error().msg()));
+ }
+ if(!output)
+ {
+ return std::unexpected(failure(output.error().msg()));
+ }
+ std::vector<std::string_view> argv;
+ for(const auto& arg : config.argv)
+ {
+ argv.push_back(arg);
+ }
+ auto process = mw::Process::execArgs(&*input, argv, &*output);
+ if(!process)
+ {
+ return std::unexpected(failure(process.error().msg()));
+ }
+ return std::unique_ptr<Transport>(new StdioTransport(
+ runtime, std::move(*input), std::move(*output), std::move(*process)));
+}
+} // namespace agent_smith::mcp::internal
diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt
index 6297fc3..c875749 100644
--- a/tests/CMakeLists.txt
+++ b/tests/CMakeLists.txt
@@ -8,6 +8,14 @@ add_executable(agent_smith_test_skeleton
test_tool.cpp
test_tool_dispatch.cpp
)
+if(AGENT_SMITH_TESTS_NEED_PYTHON)
+ find_package(Python3 REQUIRED COMPONENTS Interpreter)
+ target_sources(agent_smith_test_skeleton PRIVATE test_mcp_jsonrpc.cpp)
+ target_compile_definitions(agent_smith_test_skeleton PRIVATE
+ AGENT_SMITH_TEST_PYTHON="${Python3_EXECUTABLE}"
+ AGENT_SMITH_MCP_FIXTURE="${CMAKE_CURRENT_SOURCE_DIR}/mcp_server_fixture.py"
+ )
+endif()
target_link_libraries(agent_smith_test_skeleton
PRIVATE
agent_smith::agent_smith
diff --git a/tests/mcp_server_fixture.py b/tests/mcp_server_fixture.py
new file mode 100644
index 0000000..57e93f3
--- /dev/null
+++ b/tests/mcp_server_fixture.py
@@ -0,0 +1,145 @@
+"""Deterministic MCP protocol peer for transport and client tests."""
+
+import importlib.util
+import json
+from pathlib import Path
+import signal
+import sys
+import time
+
+spec = importlib.util.spec_from_file_location(
+ "example", Path(__file__).parent.parent / "demo/mcp_server_example.py")
+example = importlib.util.module_from_spec(spec)
+spec.loader.exec_module(example)
+mode = sys.argv[1] if len(sys.argv) > 1 else "normal"
+initialized = False
+cancelled = []
+held = None
+
+
+def emit(frame):
+ # Deliberate byte-fragmentation exercises line accumulation.
+ text = json.dumps(frame) + "\n"
+ sys.stdout.write(text[:7])
+ sys.stdout.flush()
+ sys.stdout.write(text[7:])
+ sys.stdout.flush()
+
+
+def reply(request, result):
+ emit({"jsonrpc": "2.0", "id": request["id"], "result": result})
+
+
+if mode == "stubborn":
+ signal.signal(signal.SIGTERM, signal.SIG_IGN)
+
+for line in sys.stdin:
+ request = json.loads(line)
+ method = request.get("method")
+ params = request.get("params", {})
+ if method == "notifications/initialized":
+ initialized = True
+ emit({"jsonrpc": "2.0", "method": "notifications/test",
+ "params": {"ready": True}})
+ emit({"jsonrpc": "2.0", "id": "server-ping", "method": "ping"})
+ emit({"jsonrpc": "2.0", "id": "unsupported",
+ "method": "sampling/createMessage", "params": {}})
+ continue
+ if method == "notifications/cancelled":
+ cancelled.append(params["requestId"])
+ # A server can finish work despite cancellation. This late response
+ # must not be delivered to a newer request or poison the connection.
+ emit({"jsonrpc": "2.0", "id": params["requestId"],
+ "result": {"content": [{"type": "text", "text": "late"}]}})
+ continue
+ if not method:
+ if request.get("id") == "server-ping":
+ assert request.get("result") == {}
+ if request.get("id") == "unsupported":
+ assert request.get("error", {}).get("code") == -32601
+ continue
+ if "id" not in request:
+ continue
+ if mode == "bad_json":
+ print("not-json", flush=True)
+ continue
+ if method != "initialize" and not initialized:
+ sys.exit(3)
+ if method == "initialize" and mode == "hang_init":
+ continue
+ if method == "initialize" and mode == "bad_version":
+ result = example.dispatch(method, params)
+ result["protocolVersion"] = "1999-01-01"
+ reply(request, result)
+ continue
+ if (mode == "optional_missing" and method in
+ ("resources/list", "prompts/list")) or (
+ mode == "no_tools" and method == "tools/list"):
+ emit({"jsonrpc": "2.0", "id": request["id"],
+ "error": {"code": -32601, "message": "Not supported"}})
+ continue
+ if mode == "optional_bad" and method == "resources/list":
+ reply(request, {"resources": "invalid"})
+ continue
+ if mode == "pagination" and method == "tools/list":
+ result = example.dispatch(method, params)
+ if "cursor" in params:
+ result["tools"][0]["name"] = "second"
+ else:
+ result["nextCursor"] = "page2"
+ reply(request, result)
+ continue
+ if mode == "cycle" and method == "tools/list":
+ reply(request, {"tools": [], "nextCursor": "same"})
+ continue
+ if method == "tools/call":
+ name = params["name"]
+ if name == "hang":
+ continue
+ if name == "disconnect":
+ sys.exit(0)
+ if name == "cancelled":
+ reply(request, {"content": [{"type": "text",
+ "text": json.dumps(cancelled)}]})
+ continue
+ if name == "rpc_error":
+ emit({"jsonrpc": "2.0", "id": request["id"],
+ "error": {"code": -32001, "message": "Remote failure",
+ "data": {"reason": "test"}}})
+ continue
+ if name == "bad_envelope":
+ emit({"id": request["id"], "result": {}})
+ continue
+ if name == "multimodal":
+ reply(request, {"content": [{"type": "image", "data": "AA=="}]})
+ continue
+ if name == "ordered":
+ if held is None:
+ held = request
+ else:
+ for item in (request, held):
+ reply(item, {"content": [{"type": "text",
+ "text": item["params"]["arguments"]["label"]}]})
+ held = None
+ continue
+ if mode == "binary" and method == "resources/read":
+ reply(request, {"contents": [{"uri": "example://help", "blob": "AA=="}]})
+ continue
+ if mode == "resource_disconnect" and method == "resources/read":
+ sys.exit(0)
+ if mode == "bad_prompt" and method == "prompts/get":
+ reply(request, {"messages": [{"role": "system",
+ "content": {"type": "text", "text": "test"}}]})
+ continue
+ try:
+ reply(request, example.dispatch(method, params))
+ if mode == "stop_reading" and method == "prompts/list":
+ while True:
+ time.sleep(1)
+ except LookupError:
+ emit({"jsonrpc": "2.0", "id": request["id"],
+ "error": {"code": -32601, "message": "Not supported"}})
+
+if mode == "stubborn":
+ while True:
+ time.sleep(1)
diff --git a/tests/test_mcp_jsonrpc.cpp b/tests/test_mcp_jsonrpc.cpp
new file mode 100644
index 0000000..8348a9e
--- /dev/null
+++ b/tests/test_mcp_jsonrpc.cpp
@@ -0,0 +1,484 @@
+#include <atomic>
+#include <thread>
+
+#include <agent_smith/agent_smith.hpp>
+#include <agent_smith/detail/gather_all.hpp>
+#include <gmock/gmock.h>
+#include <gtest/gtest.h>
+#include <mw/http_client_async_mock.hpp>
+
+namespace agent_smith
+{
+namespace
+{
+using namespace std::chrono_literals;
+using Json = nlohmann::json;
+
+mcp::ServerConfig server(std::string mode = "normal",
+ std::string alias = "test")
+{
+ return {.alias = std::move(alias),
+ .transport = mcp::StdioServerConfig{{AGENT_SMITH_TEST_PYTHON,
+ AGENT_SMITH_MCP_FIXTURE,
+ std::move(mode)}},
+ .request_timeout = 2s};
+}
+
+TEST(McpJsonrpcTest, InitializesDiscoversAndRendersText)
+{
+ Runtime runtime;
+ std::atomic<int> notifications{0};
+ auto result = syncWait(mcp::Client::connect(
+ runtime, server(),
+ [¬ifications](const mcp::Event& event)
+ {
+ if(std::holds_alternative<mcp::Notification>(event))
+ {
+ ++notifications;
+ }
+ }));
+ ASSERT_TRUE(result) << result.error().msg();
+ auto& client = **result;
+ EXPECT_EQ(client.info().name, "agent-smith-example");
+ ASSERT_EQ(client.tools().size(), 1U);
+ EXPECT_EQ(client.tools()[0].name, "add");
+ ASSERT_EQ(client.resources().size(), 1U);
+ EXPECT_EQ(client.resources()[0].mime_type, "text/plain");
+ ASSERT_EQ(client.prompts().size(), 1U);
+ auto sum = syncWait(client.callTool("add", {{"a", 2}, {"b", 3}}, 1s));
+ ASSERT_TRUE(sum) << sum.error().msg();
+ EXPECT_EQ(sum->content, "5");
+ EXPECT_EQ(sum->structured["sum"], 5);
+ auto resource = syncWait(client.readResource("example://help"));
+ ASSERT_TRUE(resource);
+ EXPECT_EQ(*resource, "The add tool adds two numbers.");
+ auto prompt = syncWait(client.getPrompt("greet", {{"name", "Alice"}}));
+ ASSERT_TRUE(prompt);
+ ASSERT_EQ(prompt->messages.size(), 1U);
+ EXPECT_EQ(std::get<UserMessage>(prompt->messages[0]).content,
+ "Say hello to Alice.");
+ EXPECT_GE(notifications.load(), 1);
+ EXPECT_TRUE(syncWait(client.shutdown()));
+ EXPECT_TRUE(syncWait(client.shutdown()));
+}
+
+TEST(McpJsonrpcTest, RoutesOutOfOrderResponsesConcurrently)
+{
+ Runtime runtime;
+ auto connected = syncWait(mcp::Client::connect(runtime, server()));
+ ASSERT_TRUE(connected);
+ std::vector<mw::Task<mw::E<ToolResult>>> tasks;
+ tasks.push_back(
+ (*connected)->callTool("ordered", {{"label", "first"}}, 1s));
+ tasks.push_back(
+ (*connected)->callTool("ordered", {{"label", "second"}}, 1s));
+ auto results = syncWait(internal::gatherAll(std::move(tasks)));
+ ASSERT_EQ(results.size(), 2U);
+ ASSERT_TRUE(results[0]);
+ ASSERT_TRUE(results[1]);
+ EXPECT_EQ(results[0]->content, "first");
+ EXPECT_EQ(results[1]->content, "second");
+}
+
+TEST(McpJsonrpcTest, DistinguishesToolFailureAndRpcError)
+{
+ Runtime runtime;
+ auto connected = syncWait(mcp::Client::connect(runtime, server()));
+ ASSERT_TRUE(connected);
+ auto failure = syncWait((*connected)->callTool("add", Json::object(), 1s));
+ ASSERT_TRUE(failure);
+ EXPECT_TRUE(failure->is_error);
+ auto remote =
+ syncWait((*connected)->callTool("rpc_error", Json::object(), 1s));
+ ASSERT_FALSE(remote);
+ const auto* error = remote.error().as<McpError>();
+ ASSERT_NE(error, nullptr);
+ EXPECT_EQ(error->rpc_code, -32001);
+ EXPECT_EQ(error->server, "test");
+ EXPECT_EQ(error->data["reason"], "test");
+}
+
+TEST(McpJsonrpcTest, TimesOutAndNotifiesCancellation)
+{
+ Runtime runtime;
+ auto connected = syncWait(mcp::Client::connect(runtime, server()));
+ ASSERT_TRUE(connected);
+ auto hung = syncWait((*connected)->callTool("hang", Json::object(), 30ms));
+ ASSERT_FALSE(hung);
+ EXPECT_NE(hung.error().as<TimeoutError>(), nullptr);
+ auto observed =
+ syncWait((*connected)->callTool("cancelled", Json::object(), 1s));
+ ASSERT_TRUE(observed);
+ EXPECT_EQ(Json::parse(observed->content).size(), 1U);
+ auto next =
+ syncWait((*connected)->callTool("add", {{"a", 1}, {"b", 4}}, 1s));
+ ASSERT_TRUE(next);
+ EXPECT_EQ(next->content, "5");
+}
+
+TEST(McpJsonrpcTest, CancelsAnInFlightRequest)
+{
+ Runtime runtime;
+ auto connected = syncWait(mcp::Client::connect(runtime, server()));
+ ASSERT_TRUE(connected);
+ CancellationSource cancel;
+ std::jthread thread(
+ [cancel]() mutable
+ {
+ std::this_thread::sleep_for(30ms);
+ cancel.cancel();
+ });
+ auto hung = syncWait(
+ (*connected)->callTool("hang", Json::object(), 1s, cancel.token()));
+ ASSERT_FALSE(hung);
+ EXPECT_NE(hung.error().as<CancelledError>(), nullptr);
+ auto observed =
+ syncWait((*connected)->callTool("cancelled", Json::object(), 1s));
+ ASSERT_TRUE(observed);
+ EXPECT_EQ(Json::parse(observed->content).size(), 1U);
+}
+
+TEST(McpJsonrpcTest, FiltersToolsWithDenyTakingPrecedence)
+{
+ Runtime runtime;
+ auto config = server("pagination");
+ config.allow_tools = {"add", "second"};
+ config.deny_tools = {"add"};
+ auto connected = syncWait(mcp::Client::connect(runtime, std::move(config)));
+ ASSERT_TRUE(connected);
+ ASSERT_EQ((*connected)->tools().size(), 1U);
+ EXPECT_EQ((*connected)->tools()[0].name, "second");
+}
+
+TEST(McpJsonrpcTest, OptionalMethodsMayBeMissingButMalformedListsFail)
+{
+ Runtime runtime;
+ auto connected =
+ syncWait(mcp::Client::connect(runtime, server("optional_missing")));
+ ASSERT_TRUE(connected);
+ EXPECT_TRUE((*connected)->resources().empty());
+ EXPECT_TRUE((*connected)->prompts().empty());
+ auto malformed =
+ syncWait(mcp::Client::connect(runtime, server("optional_bad")));
+ ASSERT_FALSE(malformed);
+ EXPECT_NE(malformed.error().as<McpError>(), nullptr);
+}
+
+TEST(McpJsonrpcTest, RequiredToolsMustBeAvailable)
+{
+ Runtime runtime;
+ EXPECT_FALSE(syncWait(mcp::Client::connect(runtime, server("no_tools"))));
+ auto config = server("no_tools");
+ config.tools_required = false;
+ auto connected = syncWait(mcp::Client::connect(runtime, std::move(config)));
+ ASSERT_TRUE(connected);
+ EXPECT_TRUE((*connected)->tools().empty());
+}
+
+TEST(McpJsonrpcTest, RejectsInvalidProtocolAndPagination)
+{
+ Runtime runtime;
+ for(const auto& mode : {"bad_version", "bad_json", "cycle"})
+ {
+ auto connected = syncWait(mcp::Client::connect(runtime, server(mode)));
+ ASSERT_FALSE(connected) << mode;
+ EXPECT_NE(connected.error().as<McpError>(), nullptr);
+ }
+}
+
+TEST(McpJsonrpcTest, InitializationHasDeadlineAndLaunchErrorsAreImmediate)
+{
+ Runtime runtime;
+ auto config = server("hang_init");
+ config.request_timeout = 100ms;
+ auto connected = syncWait(mcp::Client::connect(runtime, std::move(config)));
+ ASSERT_FALSE(connected);
+ EXPECT_NE(connected.error().as<TimeoutError>(), nullptr);
+ config = server();
+ config.transport = mcp::StdioServerConfig{{"/no/such/mcp/executable"}};
+ auto failed = syncWait(mcp::Client::connect(runtime, std::move(config)));
+ ASSERT_FALSE(failed);
+ ASSERT_NE(failed.error().as<McpError>(), nullptr);
+ EXPECT_TRUE(failed.error().as<McpError>()->transport);
+}
+
+TEST(McpJsonrpcTest, RejectsUnrepresentableContentAndMalformedEnvelopes)
+{
+ Runtime runtime;
+ auto connected = syncWait(mcp::Client::connect(runtime, server("binary")));
+ ASSERT_TRUE(connected);
+ EXPECT_FALSE(syncWait((*connected)->readResource("example://help")));
+ EXPECT_FALSE(
+ syncWait((*connected)->callTool("multimodal", Json::object(), 1s)));
+ EXPECT_FALSE(
+ syncWait((*connected)->callTool("bad_envelope", Json::object(), 1s)));
+ auto prompt = syncWait(mcp::Client::connect(runtime, server("bad_prompt")));
+ ASSERT_TRUE(prompt);
+ EXPECT_FALSE(syncWait((*prompt)->getPrompt("greet", Json::object())));
+}
+
+TEST(McpJsonrpcTest, DoesNotReplayWrittenToolCallsAndReconnectsNextOperation)
+{
+ Runtime runtime;
+ std::atomic<int> reconnects{0};
+ auto connected = syncWait(mcp::Client::connect(
+ runtime, server(),
+ [&reconnects](const mcp::Event& event)
+ {
+ const auto* transition = std::get_if<mcp::LifecycleEvent>(&event);
+ if(transition &&
+ transition->kind == mcp::LifecycleEventKind::RECONNECTING)
+ {
+ ++reconnects;
+ }
+ }));
+ ASSERT_TRUE(connected);
+ auto failed =
+ syncWait((*connected)->callTool("disconnect", Json::object(), 1s));
+ ASSERT_FALSE(failed);
+ EXPECT_EQ(reconnects.load(), 0);
+ auto next =
+ syncWait((*connected)->callTool("add", {{"a", 1}, {"b", 2}}, 1s));
+ ASSERT_TRUE(next) << next.error().msg();
+ EXPECT_EQ(next->content, "3");
+ EXPECT_EQ(reconnects.load(), 1);
+}
+
+TEST(McpJsonrpcTest, PrefixesAdaptersAndRetainsClientsAcrossServers)
+{
+ Runtime runtime;
+ ToolRegistry registry;
+ for(const auto& alias : {"one", "two"})
+ {
+ auto connected =
+ syncWait(mcp::Client::connect(runtime, server("normal", alias)));
+ ASSERT_TRUE(connected);
+ auto client = std::shared_ptr<mcp::Client>(std::move(*connected));
+ for(auto& tool : mcp::adaptTools(client))
+ {
+ EXPECT_TRUE(registry.add(std::move(tool)));
+ }
+ }
+ EXPECT_EQ(registry.size(), 4U);
+ ASSERT_NE(registry.find("mcp_one__add"), nullptr);
+ auto result = syncWait(registry.find("mcp_one__add")
+ ->invoke({{"a", 8}, {"b", 9}},
+ {.budget = 1s, .runtime = &runtime}));
+ ASSERT_TRUE(result);
+ EXPECT_EQ(result->content, "17");
+ auto resource = syncWait(registry.find("mcp_two__read_resource")
+ ->invoke({{"uri", "example://help"}},
+ {.budget = 1s, .runtime = &runtime}));
+ ASSERT_TRUE(resource);
+ EXPECT_EQ(resource->content, "The add tool adds two numbers.");
+}
+
+TEST(McpJsonrpcTest, ShutdownReapsUncooperativeServer)
+{
+ Runtime runtime;
+ auto connected =
+ syncWait(mcp::Client::connect(runtime, server("stubborn")));
+ ASSERT_TRUE(connected);
+ const auto start = std::chrono::steady_clock::now();
+ EXPECT_TRUE(syncWait((*connected)->shutdown()));
+ EXPECT_LT(std::chrono::steady_clock::now() - start, 2s);
+ EXPECT_FALSE(
+ syncWait((*connected)->callTool("add", {{"a", 1}, {"b", 2}}, 1s)));
+}
+
+TEST(McpJsonrpcTest, ReconnectsReadOnlyOperationsAtMostOnce)
+{
+ Runtime runtime;
+ std::atomic<int> reconnects{0};
+ auto connected = syncWait(mcp::Client::connect(
+ runtime, server("resource_disconnect"),
+ [&reconnects](const mcp::Event& event)
+ {
+ const auto* transition = std::get_if<mcp::LifecycleEvent>(&event);
+ if(transition &&
+ transition->kind == mcp::LifecycleEventKind::RECONNECTING)
+ {
+ ++reconnects;
+ }
+ }));
+ ASSERT_TRUE(connected);
+ EXPECT_FALSE(syncWait((*connected)->readResource("example://help")));
+ EXPECT_EQ(reconnects.load(), 1);
+}
+
+TEST(McpJsonrpcTest, CanDisableReconnectAndCancelBeforeSending)
+{
+ Runtime runtime;
+ auto config = server();
+ config.reconnect_on_transport_error = false;
+ auto connected = syncWait(mcp::Client::connect(runtime, std::move(config)));
+ ASSERT_TRUE(connected);
+ CancellationSource cancel;
+ cancel.cancel();
+ auto cancelled = syncWait(
+ (*connected)
+ ->callTool("add", {{"a", 1}, {"b", 2}}, 1s, cancel.token()));
+ ASSERT_FALSE(cancelled);
+ EXPECT_NE(cancelled.error().as<CancelledError>(), nullptr);
+ EXPECT_FALSE(
+ syncWait((*connected)->callTool("disconnect", Json::object(), 1s)));
+ EXPECT_FALSE(
+ syncWait((*connected)->callTool("add", {{"a", 1}, {"b", 2}}, 1s)));
+}
+
+TEST(McpJsonrpcTest, AgentUsesMcpSchemasAndRecordsToolResults)
+{
+ using ::testing::_;
+ using ::testing::Invoke;
+ Runtime runtime;
+ mw::HTTPSessionAsyncMock http;
+ LlmClient client(runtime, http,
+ {.endpoint = *mw::URL::fromStr("https://example.test"),
+ .model = "test",
+ .max_retries = 0});
+ auto memory = std::make_shared<FullMemory>();
+ Agent agent(runtime, client, memory, {.stream_default = false});
+ std::atomic<int> connected_events{0};
+ agent.hooks().on_mcp_event = [&connected_events](const mcp::Event& event)
+ {
+ if(const auto* transition = std::get_if<mcp::LifecycleEvent>(&event);
+ transition && transition->kind == mcp::LifecycleEventKind::CONNECTED)
+ {
+ ++connected_events;
+ }
+ };
+ ASSERT_TRUE(syncWait(agent.connectMcp(server())));
+ EXPECT_EQ(connected_events.load(), 1);
+ int requests = 0;
+ EXPECT_CALL(http, post(_))
+ .Times(2)
+ .WillRepeatedly(Invoke(
+ [&requests](mw::HTTPRequest request)
+ {
+ const auto body = Json::parse(request.request_data);
+ EXPECT_EQ(body["tools"][0]["function"]["name"],
+ "mcp_test__add");
+ EXPECT_EQ(body["tools"][1]["function"]["name"],
+ "mcp_test__read_resource");
+ Json message{{"role", "assistant"}, {"content", "sum is 5"}};
+ if(++requests == 1)
+ {
+ message["tool_calls"] =
+ Json::array({{{"id", "add-call"},
+ {"type", "function"},
+ {"function",
+ {{"name", "mcp_test__add"},
+ {"arguments", R"({"a":2,"b":3})"}}}}});
+ }
+ else
+ {
+ EXPECT_EQ(body["messages"].back()["role"], "tool");
+ EXPECT_EQ(body["messages"].back()["content"], "5");
+ }
+ return mw::HTTPSessionAsyncMock::complete(mw::HTTPResponse(
+ 200,
+ Json{{"choices", Json::array({{{"message", message}}})}}
+ .dump()));
+ }));
+ auto result = syncWait(agent.send("add two and three"));
+ ASSERT_TRUE(result) << result.error().msg();
+ EXPECT_EQ(result->content, "sum is 5");
+ EXPECT_EQ(memory->size(), 4U);
+}
+
+TEST(McpJsonrpcTest, AgentRejectsCollisionsWithoutPartialRegistration)
+{
+ Runtime runtime;
+ LlmClient client(runtime,
+ {.endpoint = *mw::URL::fromStr("https://example.test"),
+ .model = "test"});
+ auto connected = syncWait(mcp::Client::connect(runtime, server()));
+ ASSERT_TRUE(connected);
+ auto remote = std::shared_ptr<mcp::Client>(std::move(*connected));
+ {
+ Agent agent(runtime, client, std::make_shared<FullMemory>());
+ EXPECT_FALSE(agent.addMcpClient(nullptr));
+ ASSERT_TRUE(agent.addTool(makeTool(
+ ToolSchema{"mcp_test__read_resource",
+ "occupied",
+ {{"type", "object"}, {"properties", Json::object()}}},
+ [](Json, ToolContext) { return "local"; })));
+ EXPECT_FALSE(agent.addMcpClient(remote));
+ // A failed batch must not leave the earlier add adapter registered.
+ EXPECT_TRUE(agent.addTool(makeTool(
+ ToolSchema{"mcp_test__add",
+ "local add",
+ {{"type", "object"}, {"properties", Json::object()}}},
+ [](Json, ToolContext) { return "local"; })));
+ }
+ {
+ Agent agent(runtime, client, std::make_shared<FullMemory>());
+ ASSERT_TRUE(agent.addMcpClient(remote));
+ EXPECT_FALSE(agent.addMcpClient(remote));
+ }
+ // Hooks bound by the destroyed agent must not retain a dangling this.
+ EXPECT_TRUE(syncWait(remote->shutdown()));
+}
+
+TEST(McpJsonrpcTest, CancelsBlockedWritesWithoutWaitingForRequestDeadline)
+{
+ Runtime runtime;
+ auto connected =
+ syncWait(mcp::Client::connect(runtime, server("stop_reading")));
+ ASSERT_TRUE(connected);
+ CancellationSource cancel;
+ std::jthread thread(
+ [cancel]() mutable
+ {
+ std::this_thread::sleep_for(30ms);
+ cancel.cancel();
+ });
+ const auto start = std::chrono::steady_clock::now();
+ auto result = syncWait(
+ (*connected)
+ ->callTool("add", {{"padding", std::string(1024 * 1024, 'x')}}, 2s,
+ cancel.token()));
+ ASSERT_FALSE(result);
+ EXPECT_NE(result.error().as<CancelledError>(), nullptr);
+ EXPECT_LT(std::chrono::steady_clock::now() - start, 1s);
+}
+
+TEST(McpJsonrpcTest, AppliesWriteDeadlineWhenServerStopsReading)
+{
+ Runtime runtime;
+ auto connected =
+ syncWait(mcp::Client::connect(runtime, server("stop_reading")));
+ ASSERT_TRUE(connected);
+ const auto start = std::chrono::steady_clock::now();
+ auto result = syncWait(
+ (*connected)
+ ->callTool("add", {{"padding", std::string(1024 * 1024, 'x')}},
+ 40ms));
+ ASSERT_FALSE(result);
+ EXPECT_NE(result.error().as<TimeoutError>(), nullptr);
+ EXPECT_LT(std::chrono::steady_clock::now() - start, 1s);
+}
+
+TEST(McpJsonrpcTest, RejectsInvalidConfigurationAndDeniedDirectCalls)
+{
+ Runtime runtime;
+ auto config = server();
+ config.alias = "bad alias";
+ EXPECT_FALSE(syncWait(mcp::Client::connect(runtime, config)));
+ config = server();
+ config.request_timeout = 0ms;
+ EXPECT_FALSE(syncWait(mcp::Client::connect(runtime, config)));
+ config = server();
+ config.transport =
+ mcp::StdioServerConfig{{"python3", std::string("bad\0arg", 7)}};
+ EXPECT_FALSE(syncWait(mcp::Client::connect(runtime, config)));
+ config = server();
+ config.deny_tools = {"add"};
+ auto connected = syncWait(mcp::Client::connect(runtime, config));
+ ASSERT_TRUE(connected);
+ EXPECT_TRUE((*connected)->tools().empty());
+ EXPECT_FALSE(
+ syncWait((*connected)->callTool("add", {{"a", 1}, {"b", 2}}, 1s)));
+}
+} // namespace
+} // namespace agent_smith