#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