mcpmoqt

README

MCP over MOQT Transport

MCP over MOQT Transport 是一个实现了 Model Context Protocol (MCP) over Media over QUIC Transport (MOQT) 的 Go 语言传输层实现。

开发说明

本项目使用 Cursor 进行快速原型开发和思路验证。使用 Cursor 编写代码的目的是为了快速验证开发思路和架构设计，不代表代码本身的开发质量情况。

我们的团队承诺将持续依据 IETF 草案独立进行标准化开发，确保代码质量、性能优化、安全性以及符合相关标准规范。后续开发将遵循标准的软件工程实践，包括但不限于：

• 完整的单元测试和集成测试
• 代码审查和质量保证流程
• 性能优化和安全性审计
• 完整的文档和示例代码
• 符合 IETF 草案规范的实现

概述

本项目实现了 draft-jennings-mcp-over-moqt-00 草案中定义的 MCP over MOQT 传输协议，将 MCP 消息映射到 MOQT 对象，实现高效的发布-订阅通信。

功能特性

• ✅ 基本的对象映射为 MOQT 的载荷
• ✅ 控制轨道（Control Tracks）实现
• ✅ 客户端和服务器端传输实现
• ✅ 会话发现机制
• ✅ 连通性测试支持

版本

当前版本: v0.1.0

安装

go get github.com/mcp-moqt/mcp-moqt-transport

使用示例

服务器端

package main

import (
    "context"
    "github.com/mcp-moqt/mcp-moqt-transport"
    "github.com/mengelbart/moqtransport"
)

func main() {
    // 创建 MOQT 会话
    session := &moqtransport.Session{
        // 配置会话
    }
    
    // 创建 MCP over MOQT 服务器传输
    transport := mcpmoqt.NewMOQTServerTransport(session)
    
    // 连接到传输
    conn, err := transport.Connect(context.Background())
    if err != nil {
        // 处理错误
    }
    
    // 使用连接进行 MCP 通信
    // ...
}

客户端

package main

import (
    "context"
    "github.com/mcp-moqt/mcp-moqt-transport"
    "github.com/mengelbart/moqtransport"
)

func main() {
    // 创建 MOQT 会话
    session := &moqtransport.Session{
        // 配置会话
    }
    
    // 创建 MCP over MOQT 客户端传输
    transport := mcpmoqt.NewMOQTClientTransport(session)
    
    // 连接到传输
    conn, err := transport.Connect(context.Background())
    if err != nil {
        // 处理错误
    }
    
    // 使用连接进行 MCP 通信
    // ...
}

测试

本地连通性测试

运行本地网络连通性测试，验证QUIC和MOQT连接是否正常：

go test -v -run TestLocalConnectivity ./...

这个测试会：

• 创建一个QUIC服务器监听器
• 客户端连接到服务器
• 建立MOQT会话
• 验证基本的网络连通性

注意：客户端连接传输可能会失败（因为v0.1.0的会话发现机制还未完全实现），但这不影响连通性测试的目的 - 它验证了QUIC和MOQT层面的连接是正常的。

运行所有测试

go test -v ./...

Docker 网络测试

docker-compose up --build

项目结构

mcp-moqt-transport/
├── transport.go      # 基础传输接口和实现
├── server.go         # 服务器端传输实现
├── client.go         # 客户端传输实现
├── handler.go        # MOQT 消息处理器
├── examples/         # 示例代码
├── tests/            # 测试文件
└── README.md         # 项目文档

开发状态

当前版本 (v0.1.0) 实现了基本的传输层功能：

• Transport 和 Connection 接口实现
• MCP 消息到 MOQT 对象的映射
• 控制轨道的基本实现
• 会话发现机制
• 完整的资源轨道支持
• 工具轨道支持
• 提示轨道支持
• 通知轨道支持
• 完整的错误处理

许可证

MIT License

贡献

欢迎提交 Issue 和 Pull Request。

Documentation

Index

• Variables
• type Connection
• type MCPHandler
  • func (h *MCPHandler) Handle(rw moqtransport.ResponseWriter, msg *moqtransport.Message)
• type MCPSubscribeHandler
  • func (h *MCPSubscribeHandler) HandleSubscribe(rw *moqtransport.SubscribeResponseWriter, msg *moqtransport.SubscribeMessage)
• type MOQTClientTransport
  • func NewMOQTClientTransport(session *moqtransport.Session) *MOQTClientTransport
  • func (t *MOQTClientTransport) Connect(ctx context.Context) (Connection, error)
• type MOQTServerTransport
  • func NewMOQTServerTransport(session *moqtransport.Session) *MOQTServerTransport
  • func (t *MOQTServerTransport) Connect(ctx context.Context) (Connection, error)
• type MOQTTransport
  • func (t *MOQTTransport) Connect(ctx context.Context) (Connection, error)
• type Transport

Variables

var ErrConnectionClosed = errors.New("connection closed")

ErrConnectionClosed is returned when sending a message to a connection that is closed or in the process of closing.

Types

type Connection

type Connection interface {
	// Read reads the next message to process off the connection.
	// Connections must allow Read to be called concurrently with Close.
	Read(context.Context) (jsonrpc.Message, error)

	// Write writes a new message to the connection.
	// Write may be called concurrently.
	Write(context.Context, jsonrpc.Message) error

	// Close closes the connection.
	// Close may be called multiple times, potentially concurrently.
	Close() error

	// SessionID returns the MCP session ID for this connection.
	SessionID() string
}

Connection is a logical bidirectional JSON-RPC connection over MOQT.

type MCPHandler

type MCPHandler struct {
	// SessionID is the MCP session identifier
	SessionID string

	// ServerConn is the server connection
	ServerConn *moqtServerConn

	// SessionIDGenerator generates new session IDs
	SessionIDGenerator func() string

	// Transport is the server transport (for session management)
	Transport *MOQTServerTransport

	// PendingSessions stores session IDs that were created via discovery
	// but don't have connections yet
	PendingSessions map[string]bool
	// contains filtered or unexported fields
}

MCPHandler handles general MOQT messages for MCP.

func (*MCPHandler) Handle

func (h *MCPHandler) Handle(rw moqtransport.ResponseWriter, msg *moqtransport.Message)

Handle implements moqtransport.Handler.

type MCPSubscribeHandler

type MCPSubscribeHandler struct {
	// Transport is the server transport that manages sessions
	Transport *MOQTServerTransport
	// contains filtered or unexported fields
}

MCPSubscribeHandler handles MOQT subscribe messages for MCP tracks.

func (*MCPSubscribeHandler) HandleSubscribe

func (h *MCPSubscribeHandler) HandleSubscribe(rw *moqtransport.SubscribeResponseWriter, msg *moqtransport.SubscribeMessage)

HandleSubscribe implements moqtransport.SubscribeHandler.

type MOQTClientTransport

type MOQTClientTransport struct {
	// Session is the underlying MOQT session
	Session *moqtransport.Session

	// SessionID is the MCP session identifier (discovered from server)
	SessionID string

	// Namespace is the MOQT namespace for MCP tracks
	Namespace []string

	// ControlTrackNamespace is the namespace for control tracks
	ControlTrackNamespace []string
	// contains filtered or unexported fields
}

MOQTClientTransport implements the client side of MCP over MOQT transport.

func NewMOQTClientTransport

func NewMOQTClientTransport(session *moqtransport.Session) *MOQTClientTransport

NewMOQTClientTransport creates a new client transport.

func (*MOQTClientTransport) Connect

func (t *MOQTClientTransport) Connect(ctx context.Context) (Connection, error)

Connect implements the Transport interface for client.

type MOQTServerTransport

type MOQTServerTransport struct {
	// Session is the underlying MOQT session
	Session *moqtransport.Session

	// SessionID is the MCP session identifier
	SessionID string

	// Namespace is the MOQT namespace for MCP tracks
	Namespace []string

	// ControlTrackNamespace is the namespace for control tracks
	ControlTrackNamespace []string

	// SessionConnections maps session IDs to their connections
	SessionConnections map[string]*moqtServerConn
	// contains filtered or unexported fields
}

MOQTServerTransport implements the server side of MCP over MOQT transport.

func NewMOQTServerTransport

func NewMOQTServerTransport(session *moqtransport.Session) *MOQTServerTransport

NewMOQTServerTransport creates a new server transport.

func (*MOQTServerTransport) Connect

func (t *MOQTServerTransport) Connect(ctx context.Context) (Connection, error)

Connect implements the Transport interface for server.

type MOQTTransport

type MOQTTransport struct {
	// Session is the underlying MOQT session
	Session *moqtransport.Session

	// SessionID is the MCP session identifier
	SessionID string

	// Namespace is the MOQT namespace for MCP tracks
	Namespace []string

	// ControlTrackNamespace is the namespace for control tracks
	ControlTrackNamespace []string
}

MOQTTransport is a transport that communicates over MOQT.

func (*MOQTTransport) Connect

func (t *MOQTTransport) Connect(ctx context.Context) (Connection, error)

Connect implements the Transport interface.

type Transport

type Transport interface {
	// Connect returns a logical JSON-RPC connection over MOQT.
	Connect(ctx context.Context) (Connection, error)
}

Transport is the interface for MCP over MOQT transport. It creates connections that communicate over MOQT sessions.