背景
mcp号称ai agent时代的USB接口,本文将从源码角度理解mcp的实现原理.
本文以github.com/modelcontextprotocol/go-sdk的v1.8.0为例
本文以streamable http为例
源码
examples/client/loadtest/main.go中
func main() {
...
构建mcp client
client := mcp.NewClient(&mcp.Implementation{Name: "mcp-client", Version: "v1.0.0"}, nil)
连接mcp服务端
cs, err := client.Connect(parentCtx, &mcp.StreamableClientTransport{Endpoint: args[0]}, nil)
...
调用工具
res, err := cs.CallTool(ctx, &mcp.CallToolParams{Name: *tool, Arguments: json.RawMessage(*jsonArgs)})
...
}
前置相关
mcp/client.go中
构建mcp client
func NewClient(impl *Implementation, options *ClientOptions) *Client {
...
支持的方法
sendMethods := make(map[string]methodInfo, len(serverMethodInfos))
maps.Copy(sendMethods, serverMethodInfos)
...
c := &Client{
impl: impl,
opts: opts,
roots: newFeatureSet(func(r *Root) string { return r.URI }),
sendingMethodHandler_: defaultSendingMethodHandler,
receivingMethodHandler_: defaultReceivingMethodHandler[*ClientSession],
sendMethods: sendMethods,
}
...
}
### Connect
连接mcp服务端获取tool等信息
func (c *Client) Connect(ctx context.Context, t Transport, opts *ClientSessionOptions) (cs *ClientSession, err error) {
和服务端建立连接
cs, err = connect(ctx, t, c, (*clientSessionState)(nil), nil, c.opts.Logger)
if err != nil {
return nil, err
}
...
如果大于2026-07-28这个版本则使用最新server/discover方式
if protocolVersion >= protocolVersion20260728 {
调用server/discover来发现服务端支持的tool等
discRes, err := c.discover(discoverCtx, cs)
return cs, nil
}
...
调用initialize rpc方法
params := &InitializeParams{
ProtocolVersion: protocolVersion,
ClientInfo: c.impl,
Capabilities: c.capabilities(protocolVersion),
}
req := &InitializeRequest{Session: cs, Params: params}
res, err := handleSend[*InitializeResult](ctx, methodInitialize, req)
if err != nil {
_ = cs.Close()
return nil, err
}
if !slices.Contains(supportedProtocolVersions, res.ProtocolVersion) {
_ = cs.Close()
return nil, unsupportedProtocolVersionError{res.ProtocolVersion}
}
cs.state.InitializeResult = res
if hc, ok := cs.mcpConn.(clientConnection); ok {
hc.sessionUpdated(cs.state)
}
调用notifications/initialized rpc方法
req2 := &initializedClientRequest{Session: cs, Params: &InitializedParams{}}
if err := handleNotify(ctx, notificationInitialized, req2); err != nil {
_ = cs.Close()
return nil, err
}
...
}
调用server/discover来发现服务端支持的tool等
func (c *Client) discover(ctx context.Context, cs *ClientSession) (*InitializeResult, error) {
...
调用server/discover rpc方法
params := &DiscoverParams{
Meta: Meta{
MetaKeyProtocolVersion: protocolVersion,
MetaKeyClientInfo: c.impl,
MetaKeyClientCapabilities: caps.toV2(),
},
}
req := &DiscoverRequest{Session: cs, Params: params}
res, err := handleSend[*DiscoverResult](ctx, methodDiscover, req)
if err != nil {
return nil, err
}
...
var negotiated string
if slices.Contains(res.SupportedVersions, protocolVersion) {
negotiated = protocolVersion
} else {
negotiated = negotiateMutuallySupportedVersion(res.SupportedVersions)
}
判断服务端是否支持当前client希望的协议版本,如果协商的版本低于2026-07-28这个版本则也一样抱错
if negotiated == "" || negotiated < protocolVersion20260728 {
// If there is no overlap, fall back to initialize so version
// negotiation can happen via the legacy path.
return nil, &jsonrpc2.WireError{
Code: CodeUnsupportedProtocolVersion,
Message: "unsupported protocol version",
}
}
var serverInfo *Implementation
if v, ok := decodeMetaValue[*Implementation](res.GetMeta(), MetaKeyServerInfo); ok {
serverInfo = v
}
return &InitializeResult{
Capabilities: res.Capabilities,
Instructions: res.Instructions,
ProtocolVersion: negotiated,
ServerInfo: serverInfo,
}, nil
}
获取方法处理器
func (cs *ClientSession) sendingMethodHandler() MethodHandler {
cs.client.mu.Lock()
defer cs.client.mu.Unlock()
return cs.client.sendingMethodHandler_
}
func defaultSendingMethodHandler(ctx context.Context, method string, req Request) (Result, error) {
获取方法信息
info, ok := req.GetSession().sendingMethodInfos()[method]
if !ok {
// This can be called from user code, with an arbitrary value for method.
return nil, jsonrpc2.ErrNotHandled
}
...
调用对应rpc方法
if err := call(ctx, req.GetSession().getConn(), method, params, res); err != nil {
return nil, err
}
return res, nil
}
mcp/transport.go中
和服务端建立连接
func connect[H handler, State any](ctx context.Context, t Transport, b binder[H, State], s State, onClose func(), logger *slog.Logger) (H, error) {
...
mcpConn, err := t.Connect(ctx)
if err != nil {
return zero, err
}
...
}
mcp/streamable.go中
获取StreamableClientTransport的连接
func (t *StreamableClientTransport) Connect(ctx context.Context) (Connection, error) {
...
conn := &streamableClientConn{
url: t.Endpoint,
client: client,
incoming: make(chan jsonrpc.Message, 10),
done: make(chan struct{}),
maxRetries: maxRetries,
strict: t.strict,
logger: ensureLogger(t.logger), // must be non-nil for safe logging
ctx: connCtx,
cancel: cancel,
failed: make(chan struct{}),
disableStandaloneSSE: t.DisableStandaloneSSE,
oauthHandler: t.OAuthHandler,
maxEventSize: t.MaxEventSize,
}
return conn, nil
...
}
以写消息为例
func (c *streamableClientConn) Write(ctx context.Context, msg jsonrpc.Message) error {
...
通过sse发送消息
doRequest := func() (*http.Request, *http.Response, error) {
req, err := http.NewRequestWithContext(ctx, http.MethodPost, c.url, bytes.NewReader(data))
if err != nil {
return nil, nil, err
}
req.Header.Set("Content-Type", "application/json")
req.Header.Set("Accept", "application/json, text/event-stream")
if err := c.setMCPHeaders(req, msg); err != nil {
return nil, nil, fmt.Errorf("%s: %w: %w", requestSummary, jsonrpc2.ErrRejected, err)
}
setStandardHeaders(ctx, req.Header, msg)
resp, err := c.client.Do(req)
if err != nil {
err = fmt.Errorf("%s: %w: %w", requestSummary, jsonrpc2.ErrRejected, err)
}
return req, resp, err
}
req, resp, err := doRequest()
if err != nil {
return err
}
...
}
CallTool
mcp/client.go中
func (cs *ClientSession) CallTool(ctx context.Context, params *CallToolParams) (*CallToolResult, error) {
调用methodCallTool rpc 方法
return handleSend[*CallToolResult](ctx, methodCallTool, newClientRequest(cs, orZero[Params](params)))
}
