Gym MCP Client

代理令

一个统一的Python客户端，用于通过以下方式处理本地和远程Gymnasium环境 健身房mcp服务器.

目标：只需编写一次代码，即可在本地开发和远程执行之间无缝切换。

特性

• 🎮 统一API：本地和远程环境的健身房界面相同
• 🔄 无缝切换：使用单个参数更改模式
• 🌐 远程执行：通过HTTP连接到健身房mcp服务器实例
• 🤖 MCP扩展：与SDK无关的工具生成，支持自动完成访问（与SDK无关）
• 🔧 完全兼容性：支持所有体育馆环境类型（盒子、离散、多二进制等）
• 🐍 现代Python：Python 3.10+，带有完整的类型提示
• 📦 轻松设置：使用uv管理，实现快速依赖管理
• 🌐 统一配置：服务器URL的环境变量
• 💡 IDE自动补全：ToolCollection提供这两种功能 tools["name"] 和 tools.name 访问
• 📊 结果分析：全面的剧集统计和导出功能
• ✅ 测试良好：19个核心测试+7个MCP扩展测试套件

安装

# Install with uv (recommended)
uv sync

# Or with pip
pip install -e .

环境变量

AgentRing支持用于配置的环境变量：

GYM_SERVER_URL

为两者设置默认的MCP服务器URL gym.make() 和 gym_mcp.create_tools():

export GYM_SERVER_URL="http://localhost:8070"

这使您能够：

• 使用 gym.make("CartPole-v1", mode="remote") 未具体说明 gym_server_url
• 使用 gym_mcp.create_tools() 未具体说明 server_url
• 通过更新一个环境变量来更改整个项目中的服务器URL

快速开始

基本用法

本地模式（标准体育馆）

import agentring as gym

# Create local environment (same as gymnasium.make)
env = gym.make("CartPole-v1", render_mode="human")

observation, info = env.reset()
action = env.action_space.sample()  # Random action
observation, reward, terminated, truncated, info = env.step(action)

env.close()

远程模式（MCP服务器）

import agentring as gym

# Create remote environment via MCP server
env = gym.make(
    "CartPole-v1",
    mode="remote",
    gym_server_url="http://localhost:8000"
)

observation, info = env.reset()
action = env.action_space.sample()
observation, reward, terminated, truncated, info = env.step(action)

env.close()

MCP扩展快速入门

AgentRing的MCP扩展极大地简化了代理开发：

环境变量配置（推荐）

为整个项目设置一次服务器URL：

export GYM_SERVER_URL="http://localhost:8070"

1.生成工具（SDK不可知）

import agentring.mcp as gym_mcp

# One line to get all environment tools (uses GYM_SERVER_URL)
tools = gym_mcp.create_tools()

# Or specify URL explicitly
tools = gym_mcp.create_tools("http://localhost:8070")

# Tools are accessed by name (dict return)
reset_result = tools["reset_env"](seed=42)  # reset_env
step_result = tools["step_env"](action="go north")  # step_env

2.与任何SDK一起使用

# With CrewAI
from crewai import Agent
from crewai.tools import tool

@tool
def reset_env(seed=None):
    return tools["reset_env"](seed=seed)

@tool
def step_env(action: str):
    return tools["step_env"](action=action)

agent = Agent(tools=[reset_env, step_env], ...)

3.SDK原生剧集执行

# Agent SDKs handle episode execution natively
# AgentRing provides tools and result collection

# Example with CrewAI
from crewai import Agent, Task, Crew

agent = Agent(tools=[reset_env, step_env, ...], ...)
task = Task(description="Complete the quest", agent=agent)
crew = Crew(agents=[agent], tasks=[task])
result = crew.kickoff()

# Collect results with AgentRing
episode_results = gym_mcp.results.EpisodeResults([
    gym_mcp.types.EpisodeResult(1, 0.8, 12, True)
])
print(episode_results.summary())

通过SDK完成示例

CrewAI+文本世界

import agentring.mcp as gym_mcp
from crewai import Agent, Task, Crew
from crewai.tools import tool

# Generate tools
tools = gym_mcp.create_tools("http://localhost:8070")

# Wrap for CrewAI
@tool
def reset_env(seed=None):
    return tools["reset_env"](seed=seed)

@tool
def step_env(action: str):
    return tools["step_env"](action=action)

# Create agent
agent = Agent(
    role="Text Adventure Agent",
    goal="Complete quests in text environments",
    backstory="You are skilled at solving puzzles and exploring.",
    tools=[reset_env, step_env],
    verbose=True
)

# Run task
task = Task(
    description="Find the treasure and escape the dungeon",
    agent=agent
)

crew = Crew(agents=[agent], tasks=[task])
result = crew.kickoff()

LangGraph+ALFWorld

import agentring.mcp as gym_mcp
from langchain_core.tools import tool
from langgraph import StateGraph, START, END
from langgraph.prebuilt import ToolNode

# Generate tools
tools = gym_mcp.create_tools("http://localhost:8090")

# Wrap for LangChain
@tool
def reset_env(seed=None):
    return tools["reset_env"](seed=seed)

@tool
def step_env(action: str):
    return tools["step_env"](action=action)

# Create LangGraph workflow
def agent_node(state):
    # Your LLM logic here
    return {"messages": state["messages"] + ["response"]}

workflow = StateGraph()
workflow.add_node("agent", agent_node)
workflow.add_node("tools", ToolNode([reset_env, step_env]))
workflow.add_edge(START, "agent")
workflow.add_edge("tools", "agent")
workflow.add_conditional_edges("agent", lambda s: END if s.get("done") else "tools")

app = workflow.compile()
result = app.invoke({"messages": ["Complete the household task"]})

谷歌ADK+WebShop

import agentring.mcp as gym_mcp
from google.adk.agents import LlmAgent
from google.adk.tools import FunctionTool

# Generate tools
tools = gym_mcp.create_tools("http://localhost:8002")

# Wrap for Google ADK
def reset_env(seed=None):
    return tools["reset_env"](seed=seed)

def step_env(action: str):
    return tools["step_env"](action=action)

reset_tool = FunctionTool(reset_env)
step_tool = FunctionTool(step_env)

# Create agent
agent = LlmAgent(
    name="ShoppingAgent",
    description="An agent that shops efficiently",
    model="gemini-2.0-flash-exp",
    instruction=gym_mcp.templates.SHOPPING_INSTRUCTIONS,
    tools=[reset_tool, step_tool],
)

# Run episode
import asyncio
async for result in agent.run_async("Buy the best laptop for under $1000"):
    print(result.text)

多服务器示例

import agentring.mcp as gym_mcp

# Connect to multiple environments
multi_client = gym_mcp.MultiServerClient()
multi_client.add_server("textworld", "http://localhost:8070")
multi_client.add_server("alfworld", "http://localhost:8090")
multi_client.add_server("webshop", "http://localhost:8002")

# Get tools from all servers
all_tools = multi_client.get_all_tools()

# Check server health
health = multi_client.health_check_all()
print(f"Server health: {health}")

# Run agents across different environments
textworld_tools = multi_client.get_tools("textworld")
alfworld_tools = multi_client.get_tools("alfworld")

# SDKs handle episode execution natively
# Use AgentRing for result collection and analysis

# Example: Run TextWorld agent
# result = your_textworld_agent.run("Complete the quest")
# Collect results with AgentRing
tw_results = gym_mcp.results.EpisodeResults([
    gym_mcp.types.EpisodeResult(1, 0.8, 12, True),
    # ... more episode results
])

aw_results = gym_mcp.results.EpisodeResults([
    gym_mcp.types.EpisodeResult(1, 0.6, 15, True),
    # ... more episode results
])

print("TextWorld:", tw_results.success_percentage, "% success")
print("ALFWorld:", aw_results.success_percentage, "% success")

建筑

┌─────────────────┐
│  Your Code      │
└────────┬────────┘
         │
         ▼
┌─────────────────┐
│ gym.make    │
└────────┬────────┘
         │
    ┌────┴────┐
    ▼         ▼
┌────────┐ ┌──────────────┐
│ Local  │ │ Remote       │
│ Gym    │ │ HTTP Client  │
└────────┘ └──────┬───────┘
              │
              ▼
         ┌──────────────┐
         │ gym-mcp-     │
         │ server       │
         └──────┬───────┘
                │
                ▼
           ┌────────┐
           │ Gym    │
           │ Env    │
           └────────┘

支持的空间类型

例子

看 quickstart.py 显示本地和远程模式的完整工作示例。

运行示例：

# Local mode (default)
uv run python quickstart.py

# Remote mode (start server first!)
python -m gym_mcp_server --env CartPole-v1 --transport streamable-http --port 8000
# Then edit quickstart.py to set REMOTE_MODE = True and run:
uv run python quickstart.py

发展

设置

git clone 
cd agentring
make install

可用命令

make help       # Show all commands
make test       # Run test suite (14 tests)
make lint       # Run ruff linter
make format     # Format code with ruff
make typecheck  # Run mypy type checker
make check      # Run all checks (lint + typecheck + test)
make all        # Format, then run all checks
make demo       # Run local demo
make clean      # Clean build artifacts

运行测试

make test
# Or: uv run pytest tests/ -v
# 19 tests (18 passing, 1 failing due to missing pygame) ✅

用例

1. 发展→ 生产：本地开发，远程部署
2. 分布式培训：多个进程连接到远程环境
3. 资源管理：在专用服务器上运行昂贵的模拟
4. 测试：在远程部署之前进行本地测试

演出

本地模式

• 开销：最小（薄包装）
• 最适合：开发、测试、轻量级环境

远程模式

• 开销：HTTP往返（本地主机上1-10ms）
• 最适合：昂贵的环境、分布式培训、资源共享

错误处理

import agentring as gym
import httpx

try:
    env = gym.make(
        "CartPole-v1",
        mode="remote",
        gym_server_url="http://localhost:8000"
    )
    observation, info = env.reset()
    # ... your code ...
except ValueError:
    # Invalid mode, missing URL, etc.
    pass
except RuntimeError:
    # Environment initialization failed, remote call failed
    pass
except httpx.HTTPError:
    # Network error (remote mode only)
    pass
finally:
    if 'env' in locals():
        env.close()

故障排除

远程连接问题

1. 确保健身房mcp服务器正在运行且可访问
2. 检查URL是否正确（包括协议： http:// 或 https://)
3. 验证防火墙/网络设置
4. 检查服务器日志是否有错误

未找到环境

# For Atari environments
uv add "gymnasium[atari]"

# For Box2D environments
uv add "gymnasium[box2d]"

# For MuJoCo environments
uv add "gymnasium[mujoco]"

需求

• Python 3.10+
• 体育馆>=1.2.1
• httpx>=0.28.1
• numpy>=2.0.0
• gym-mcp服务器（来自GitHub）

贡献

欢迎投稿！拜托：

1. 克隆该仓库
2. 创建要素分支
3. 进行更改
4. 跑 make check 验证所有测试是否通过
5. 提交拉取请求

看 CONTRIBUTING.md 详细指南。

许可证

MIT许可证-有关详细信息，请参阅许可证文件。

MCP代理开发扩展

AgentRing现在包括强大的MCP（模型上下文协议）扩展，大大简化了MCP服务器的代理开发。这些扩展提供通用的、与SDK无关的工具和实用程序。

特性

• 🤖 通用工具厂：从MCP服务器自动生成可调用工具（适用于任何代理SDK）
• 🎯 SDK不可知：没有SDK依赖关系-工具是标准的Python可调用工具
• 🚀 剧集跑者：统一的事件执行和结果收集
• 🔧 格式转换器：将工具定义转换为JSON模式、OpenAPI和SDK特定格式
• 🌐 多服务器支持：同时使用多个MCP服务器
• 📊 结果分析：全面的剧集结果统计和导出功能

快速开始

import agentring.mcp as gym_mcp

# 1. Generate tools from MCP server
tools = gym_mcp.create_tools("http://localhost:8070")
# Returns: List of callable Python functions

# 2. Use with any agent SDK
# Example with CrewAI:
from crewai import Agent, Task, Crew
from crewai.tools import tool

@tool
def reset_env(seed=None):
    return tools["reset_env"](seed=seed)

@tool
def step_env(action):
    return tools["step_env"](action=action)

agent = Agent(tools=[reset_env, step_env], ...)
task = Task(description="Complete the household task", agent=agent)
crew = Crew(agents=[agent], tasks=[task])
result = crew.kickoff()

# 3. SDKs handle episode execution natively
# AgentRing provides tools and result collection
# Use AgentRing's EpisodeResults for analysis:
results = gym_mcp.results.EpisodeResults([
    gym_mcp.types.EpisodeResult(1, 0.8, 12, True),
    # ... collect results from SDK execution
])
print(results.summary())

MCP工具厂

这 create_tools() 函数自动从MCP服务器发现可用工具并返回 ToolCollection 它通过IDE自动补全功能提供字典样式和属性样式访问：

# Generate all tools from server
tools = gym_mcp.create_tools("http://localhost:8070")

# Generate specific tools
tools = gym_mcp.create_tools("http://localhost:8070", ["reset_env", "step_env"])

# Tools support both dict-style and attribute-style access
result = tools["reset_env"](seed=42)  # Dict-style access
result = tools.reset_env(seed=42)     # Attribute access with autocomplete!

# Both access methods provide the same callable function
assert tools["reset_env"] is tools.reset_env  # True

SDK原生剧集执行

代理SDK以本机方式处理剧集执行。AgentRing提供工具和结果收集：

# Example with CrewAI
from crewai import Agent, Task, Crew

agent = Agent(tools=[reset_env, step_env], ...)
task = Task(description="Complete the task", agent=agent)
crew = Crew(agents=[agent], tasks=[task])
result = crew.kickoff()  # SDK handles execution

# Use AgentRing for result collection and analysis
from agentring.mcp import results, types

episode_results = results.EpisodeResults([
    types.EpisodeResult(1, 0.8, 12, True),
    # ... collect from actual runs
])
print(f"Success rate: {episode_results.success_percentage:.1f}%")

格式转换器

将工具定义转换为各种格式以进行SDK集成：

from agentring.mcp import formats

# Convert to JSON Schema (OpenAI style)
json_schema = formats.to_json_schema(tool_definition)

# Convert to OpenAPI spec
openapi_spec = formats.to_openapi_spec(tool_definition)

# Convert to SDK-specific formats
crewai_format = formats.to_crewai_tool(tool_definition)
langchain_format = formats.to_langchain_tool(tool_definition)

多服务器支持

使用多个MCP服务器：

multi_client = gym_mcp.MultiServerClient()
multi_client.add_server("textworld", "http://localhost:8070")
multi_client.add_server("alfworld", "http://localhost:8090")

# Get tools from all servers
all_tools = multi_client.get_all_tools()

# Health check all servers
health = multi_client.health_check_all()
print(f"Healthy servers: {multi_client.get_healthy_servers()}")

代理模板

常见代理模式的预构建指令模板：

from agentring.mcp import templates

# Get templates
text_adventure_prompt = templates.TEXT_ADVENTURE_INSTRUCTIONS
shopping_prompt = templates.SHOPPING_INSTRUCTIONS
household_prompt = templates.HOUSEHOLD_INSTRUCTIONS

# Create complete agent configurations
config = templates.create_text_adventure_config(
    max_steps=50,
    custom_instructions="Always examine objects before using them."
)

结果分析

综合事件结果分析和导出：

results = runner.run_episodes(episodes=20)

# Statistics
print(f"Success rate: {results.success_percentage:.1f}%")
print(f"Average reward: {results.average_reward:.2f}")
print(f"Average steps: {results.average_steps:.1f}")

# Export results
results.save_json("results.json")
results.save_csv("results.csv")

# Filter and analyze
successful_episodes = results.filter_by_success(successful_only=True)
high_reward_episodes = results.filter_by_reward(min_reward=1.0)

SDK集成示例

与CrewAI合作

import agentring.mcp as gym_mcp
from crewai import Agent, Task, Crew
from crewai.tools import tool

# Generate tools
tools = gym_mcp.create_tools("http://localhost:8070")

# Wrap for CrewAI
@tool
def reset_env(seed=None):
    return tools["reset_env"](seed=seed)

@tool
def step_env(action):
    return tools["step_env"](action=action)

agent = Agent(
    role="Text Adventure Agent",
    goal="Complete quests in text worlds",
    backstory="You excel at solving puzzles and exploring environments.",
    tools=[reset_env, step_env]
)

task = Task(description="Find the treasure and escape the dungeon", agent=agent)
crew = Crew(agents=[agent], tasks=[task])
result = crew.kickoff()

使用LangGraph

import agentring.mcp as gym_mcp
from langchain_core.tools import tool
from langgraph import StateGraph

# Generate and wrap tools
tools = gym_mcp.create_tools("http://localhost:8070")

@tool
def reset_env(seed=None):
    return tools["reset_env"](seed=seed)

@tool
def step_env(action):
    return tools["step_env"](action=action)

# Use in LangGraph workflow
# ... workflow definition ...

使用通用代理

import agentring.mcp as gym_mcp

# Generate tools
tools = gym_mcp.create_tools("http://localhost:8070")

# Define custom agent class
class MyCustomAgent:
    def __init__(self, tools):
        self.tools = tools
        self.episode_results = []

    def run_episode(self, prompt: str, max_steps=10):
        # Custom episode execution logic
        # Use tools["reset_env"](), tools["step_env"](), etc.
        total_reward = 0.0
        steps = 0

        # Reset environment
        reset_result = self.tools["reset_env"](seed=42)

        # Your agent logic here
        for step in range(max_steps):
            # Agent decision making
            action = "look"  # Your logic here

            # Execute action
            step_result = self.tools["step_env"](action=action)
            reward = step_result.get("reward", 0)
            total_reward += reward
            steps += 1

            if step_result.get("done"):
                break

        # Store result
        result = gym_mcp.types.EpisodeResult(1, total_reward, steps, total_reward > 0)
        self.episode_results.append(result)
        return result

    def get_results(self):
        return gym_mcp.results.EpisodeResults(self.episode_results)

# Use the custom agent
agent = MyCustomAgent(tools)
agent.run_episode("Complete the task")
results = agent.get_results()

SDK集成指南

AgentRing的MCP扩展适用于任何代理SDK。以下是流行框架的综合示例：

CrewAI集成

CrewAI代理可以使用AgentRing MCP工具，只需进行最少的代码更改。

基本CrewAI代理

import agentring.mcp as gym_mcp
from crewai import Agent, Task, Crew
from crewai.tools import tool

# 1. Generate tools from MCP server
tools = gym_mcp.create_tools("http://localhost:8070")

# 2. Wrap tools for CrewAI (simple adapters)
@tool
def reset_env(seed=None):
    """Reset the environment to start a new episode."""
    return tools["reset_env"](seed=seed)

@tool
def step_env(action: str):
    """Take an action in the environment."""
    return tools["step_env"](action=action)

@tool
def get_env_info():
    """Get information about the environment."""
    return tools["get_env_info"]()

# 3. Create CrewAI agent
agent = Agent(
    role="Text Adventure Agent",
    goal="Complete quests and solve puzzles in text-based environments",
    backstory="""You are an expert at playing text adventure games.
    You carefully read descriptions, make logical decisions, and
    systematically explore environments to achieve objectives.""",
    tools=[reset_env, step_env, get_env_info],
    verbose=True,
    allow_delegation=False
)

# 4. Create and run task
task = Task(
    description="""Navigate the environment, find the treasure,
    and return it to the starting location. Be methodical and
    examine objects before using them.""",
    agent=agent,
    expected_output="A summary of the completed quest"
)

crew = Crew(agents=[agent], tasks=[task], verbose=True)
result = crew.kickoff()

具有多个代理的高级CrewAI

import agentring.mcp as gym_mcp
from crewai import Agent, Task, Crew
from crewai.tools import tool

# Generate tools from multiple servers
textworld_tools = gym_mcp.create_tools("http://localhost:8070")
alfworld_tools = gym_mcp.create_tools("http://localhost:8090")

# Wrap tools
@tool
def reset_textworld(seed=None):
    return textworld_tools["reset_env"](seed=seed)

@tool
def step_textworld(action: str):
    return textworld_tools["step_env"](action=action)

@tool
def reset_alfworld(seed=None):
    return alfworld_tools["reset_env"](seed=seed)

@tool
def step_alfworld(action: str):
    return alfworld_tools["step_env"](action=action)

# Create specialized agents
textworld_agent = Agent(
    role="Text Adventure Expert",
    goal="Solve text-based puzzles and quests",
    tools=[reset_textworld, step_textworld],
    verbose=True
)

alfworld_agent = Agent(
    role="Household Task Expert",
    goal="Complete household tasks efficiently",
    tools=[reset_alfworld, step_alfworld],
    verbose=True
)

# Create crew with multiple agents
crew = Crew(
    agents=[textworld_agent, alfworld_agent],
    tasks=[
        Task(description="Solve the treasure quest", agent=textworld_agent),
        Task(description="Clean the kitchen", agent=alfworld_agent)
    ],
    verbose=True
)

result = crew.kickoff()

LangGraph集成

LangGraph工作流可以将AgentRing MCP工具用作LangChain工具。

基本语言图代理

import agentring.mcp as gym_mcp
from langchain_core.tools import tool
from langgraph import StateGraph, START, END
from langgraph.prebuilt import ToolNode
from typing import TypedDict

# 1. Generate tools from MCP server
tools = gym_mcp.create_tools("http://localhost:8070")

# 2. Wrap tools for LangChain
@tool
def reset_env(seed: int = None) -> str:
    """Reset the environment to start a new episode."""
    result = tools["reset_env"](seed=seed)
    return f"Environment reset: {result}"

@tool
def step_env(action: str) -> str:
    """Take an action in the environment."""
    result = tools["step_env"](action=action)
    return f"Action result: {result}"

# 3. Define state
class AgentState(TypedDict):
    messages: list
    step_count: int
    total_reward: float
    done: bool

# 4. Create workflow
def agent_node(state: AgentState):
    # Agent logic here (LLM call with tools)
    messages = state["messages"]
    # ... LLM call with tool calling ...
    return {"messages": messages, "step_count": state["step_count"] + 1}

def should_continue(state: AgentState) -> str:
    if state["done"] or state["step_count"] >= 50:
        return END
    return "tools"

# 5. Build graph
workflow = StateGraph(AgentState)
workflow.add_node("agent", agent_node)
workflow.add_node("tools", ToolNode([reset_env, step_env]))

workflow.add_edge(START, "agent")
workflow.add_edge("tools", "agent")
workflow.add_conditional_edges("agent", should_continue)

app = workflow.compile()

# 6. Run workflow
initial_state = {
    "messages": [{"role": "user", "content": "Complete the text adventure quest"}],
    "step_count": 0,
    "total_reward": 0.0,
    "done": False
}

result = app.invoke(initial_state)
print(f"Workflow completed with {result['step_count']} steps")

带有代理程序的LangGraph MCP运行器

import agentring.mcp as gym_mcp
import asyncio
from langchain_core.language_models import BaseLanguageModel

# 1. Generate tools
tools = gym_mcp.create_tools("http://localhost:8070")

# 2. Create LangGraph agent interface
async def langgraph_agent(prompt: str) -> str:
    # Your LangGraph agent logic here
    # This would integrate with your LangGraph setup
    return "Agent response with tool calls"

# 3. Integrate with LangGraph workflow
# Add tools to your LangGraph workflow nodes
# Your LangGraph handles episode execution natively

# 4. Collect results
results = gym_mcp.results.EpisodeResults([
    gym_mcp.types.EpisodeResult(1, 0.85, 14, True),
    # ... collect from LangGraph execution
])
print(results.summary())

谷歌ADK集成

Google ADK（代理开发工具包）与AgentRing MCP工具无缝协作。

基本的Google ADK代理

import agentring.mcp as gym_mcp
from google.adk.agents import LlmAgent
from google.adk.tools import FunctionTool

# 1. Generate tools from MCP server
tools = gym_mcp.create_tools("http://localhost:8070")

# 2. Wrap tools for Google ADK
def reset_env_adk(seed=None):
    """Reset the environment to start a new episode."""
    return tools["reset_env"](seed=seed)

def step_env_adk(action: str):
    """Take an action in the environment."""
    return tools["step_env"](action=action)

def get_env_info_adk():
    """Get information about the environment."""
    return tools["get_env_info"]()

# Create FunctionTool instances
reset_tool = FunctionTool(reset_env_adk)
step_tool = FunctionTool(step_env_adk)
info_tool = FunctionTool(get_env_info_adk)

# 3. Create ADK agent
agent = LlmAgent(
    name="TextWorldAgent",
    description="An agent that plays TextWorld text adventure games",
    model="gemini-2.0-flash-exp",
    instruction=gym_mcp.templates.TEXT_ADVENTURE_INSTRUCTIONS,
    tools=[reset_tool, step_tool, info_tool],
)

# 4. Run episodes (async)
async def run_adk_episodes():
    results = []
    for episode in range(3):
        result = await agent.run_async(f"Episode {episode + 1}: Complete the quest")
        results.append(result)
    return results

# Run the episodes
import asyncio
episode_results = asyncio.run(run_adk_episodes())

带有AgentRing Runner的高级Google ADK

import agentring.mcp as gym_mcp
import asyncio

# 1. Generate tools and create ADK agent
tools = gym_mcp.create_tools("http://localhost:8070")

# Create ADK agent (same as above)
agent = LlmAgent(
    name="TextWorldAgent",
    description="An agent that plays TextWorld text adventure games",
    model="gemini-2.0-flash-exp",
    instruction=gym_mcp.templates.TEXT_ADVENTURE_INSTRUCTIONS,
    tools=[reset_tool, step_tool, info_tool],
)

# 2. Create ADK agent interface for AgentRing runner
async def adk_agent_interface(prompt: str) -> str:
    """Adapter to use ADK agent with AgentRing runner."""
    # Handle ADK's async generator response
    async for result in agent.run_async(prompt):
        return result.text
    return "No response from agent"

# 3. Run episodes (ADK handles execution natively)
# ADK agents execute episodes using their built-in run_async method
# with integrated tool calling

# 4. Collect results for analysis
results = gym_mcp.results.EpisodeResults([
    gym_mcp.types.EpisodeResult(1, 0.82, 16, True),
    # ... collect from ADK execution
])
print(results.summary())

OpenAI代理SDK集成

OpenAI代理SDK具有原生MCP支持，使集成更加简单。

使用MCP的基本OpenAI代理

from agents import Agent, Runner, ModelSettings
from agents.mcp import MCPServerStreamableHttp

# 1. Create MCP server connection (no AgentRing needed for basic usage)
server = MCPServerStreamableHttp(
    name="Gym Environment",
    params={"url": "http://localhost:8070/mcp", "timeout": 10},
)

# 2. Create agent with MCP server
agent = Agent(
    name="GymAgent",
    instructions="You are an agent playing in a Gym environment. Complete tasks efficiently.",
    mcp_servers=[server],
    model="gpt-4o",
    model_settings=ModelSettings(temperature=0.1),
)

# 3. Run agent
result = await Runner.run(agent, "Complete the text adventure quest")
print(result.final_output)

带有代理路由MCP扩展的OpenAI代理

import agentring.mcp as gym_mcp
from agents import Agent, Runner, ModelSettings

# 1. Use AgentRing to get MCP server connection
client = gym_mcp.MCPServerClient("http://localhost:8070")
server = MCPServerStreamableHttp(
    name="Gym Environment",
    params={"url": f"{client.server_url}/mcp", "timeout": 10},
)

# 2. Create agent
agent = Agent(
    name="GymAgent",
    instructions=gym_mcp.templates.TEXT_ADVENTURE_INSTRUCTIONS,
    mcp_servers=[server],
    model="gpt-4o",
    model_settings=ModelSettings(temperature=0.1),
)

# 3. Run episodes (OpenAI Agents SDK handles execution natively)
# The OpenAI Agents SDK manages episode execution with built-in tool calling
async def run_openai_agent(prompt: str):
    result = await Runner.run(agent, prompt)
    return result.final_output

# Use AgentRing for result collection and analysis
results = gym_mcp.results.EpisodeResults([
    gym_mcp.types.EpisodeResult(1, 0.88, 13, True),
    # ... collect from OpenAI Agents execution
])

Letta集成

Letta代理可以使用AgentRing MCP工具作为函数定义。

基础Letta代理

import agentring.mcp as gym_mcp
from letta_client import Letta

# 1. Generate tools from MCP server
tools = gym_mcp.create_tools("http://localhost:8070")

# 2. Convert to Letta tool format
letta_tools = []
for tool in tools:
    tool_def = gym_mcp.formats.to_letta_tool(tool)
    letta_tools.append(tool_def)

# 3. Create Letta client
client = Letta(api_key="your-letta-api-key")

# 4. Create Letta agent with tools
agent_state = client.agents.create(
    model="openai/gpt-4o-mini",
    embedding="openai/text-embedding-3-small",
    memory_blocks=[
        {
            "label": "persona",
            "value": gym_mcp.templates.TEXT_ADVENTURE_INSTRUCTIONS
        }
    ],
    # Letta handles tool registration differently
)

# 5. Run agent
response = client.agents.messages.create(
    agent_id=agent_state.id,
    input="Start a new text adventure episode and complete the quest"
)

自定义SDK集成

对于任何自定义或不受支持的SDK，您可以直接使用AgentRing的通用工具。

具有自定义SDK的通用代理

import agentring.mcp as gym_mcp

# 1. Generate tools from MCP server
tools = gym_mcp.create_tools("http://localhost:8070")

# 2. Use tools directly in your custom agent
class MyCustomAgent:
    def __init__(self, tools):
        self.tools = tools

    def run_episode(self, instructions: str):
        # Reset environment
        reset_result = self.tools["reset_env"](seed=42)
        print(f"Environment reset: {reset_result}")

        # Your custom agent logic here
        # Use self.tools["step_env"] for step_env, etc.

        return {"success": True, "steps": 5, "reward": 1.0}

# 3. Create and run agent
agent = MyCustomAgent(tools)
result = agent.run_episode("Complete the text adventure quest")

多SDK代理系统

您甚至可以创建使用多个SDK执行不同任务的代理。

多SDK系统

import agentring.mcp as gym_mcp

# 1. Set up multiple MCP servers
multi_client = gym_mcp.MultiServerClient()
multi_client.add_server("textworld", "http://localhost:8070")
multi_client.add_server("alfworld", "http://localhost:8090")
multi_client.add_server("webshop", "http://localhost:8002")

# 2. Get tools from all servers
all_tools = multi_client.get_all_tools()

# 3. Create agents for different domains
textworld_tools = multi_client.get_tools("textworld")
alfworld_tools = multi_client.get_tools("alfworld")
webshop_tools = multi_client.get_tools("webshop")

# 4. Use different SDKs for different environments
def create_specialized_agent(sdk_name: str, tools: list, env_name: str):
    """Create an agent using the specified SDK for a specific environment."""

    if sdk_name == "crewai":
        from crewai import Agent
        from crewai.tools import tool

        # Wrap tools for CrewAI
        @tool
        def reset_env(seed=None):
            return tools["reset_env"](seed=seed)

        @tool
        def step_env(action: str):
            return tools["step_env"](action=action)

        return Agent(
            role=f"{env_name} Specialist",
            goal=f"Excel at tasks in {env_name}",
            tools=[reset_env, step_env],
            verbose=True
        )

    elif sdk_name == "langgraph":
        # LangGraph implementation
        pass

    # Add other SDKs...

    else:
        raise ValueError(f"Unsupported SDK: {sdk_name}")

# 5. Create specialized agents
textworld_agent = create_specialized_agent("crewai", textworld_tools, "TextWorld")
alfworld_agent = create_specialized_agent("crewai", alfworld_tools, "ALFWorld")

# 6. Run agents on their respective environments (SDK-native execution)
# textworld_result = textworld_agent.run("Complete TextWorld quest")
# alfworld_result = alfworld_agent.run("Complete ALFWorld task")

# 7. Collect and compare performance with AgentRing results
textworld_results = gym_mcp.results.EpisodeResults([
    gym_mcp.types.EpisodeResult(1, 0.85, 14, True),
    # ... collect results from actual runs
])

alfworld_results = gym_mcp.results.EpisodeResults([
    gym_mcp.types.EpisodeResult(1, 0.72, 18, True),
    # ... collect results from actual runs
])

print("TextWorld Results:")
print(textworld_results.summary())
print("\nALFWorld Results:")
print(alfworld_results.summary())

api参考

API核心代理

agentring.make(id, mode="local", **kwargs)

营造一个体育馆环境。

参数：

• id （str）：环境ID（例如“CartPole-v1”）
• mode （str）：“本地”或“远程”
• render_mode （str，可选）：渲染模式
• gym_server_url （str，可选）：远程模式的MCP服务器URL
• **kwargs：其他论点

退货： AgentRingClient实例

MCP扩展API

agentring.mcp.create_tools(server_url, tool_names=None, client=None)

从MCP服务器创建可调用工具。

参数：

• server_url （str）：MCP服务器URL
• tool_names （列表，可选）：要创建的特定工具名称
• client （MCPServerClient，可选）：预配置的客户端

退货： ToolCollection，其中包含可按名称访问的工具(tools["reset_env"])或属性(tools.reset_env)用于IDE自动补全

agentring.mcp.MCPServerClient(server_url, **kwargs)

带连接管理的MCP服务器客户端。

参数：

• server_url （str）：服务器URL
• timeout （float）：请求超时
• max_retries （int）：最大重试次数
• health_check_interval （浮动）：健康检查间隔

方法：

• health_check()：检查服务器运行状况
• call_tool(tool_name, params)：调用工具
• get_server_info()：获取服务器信息

agentring.mcp.MultiServerClient()

管理多个MCP服务器。

方法：

• add_server(name, url)：添加服务器
• get_tools(server_name)：从服务器获取工具
• get_all_tools()：从所有服务器获取工具
• health_check_all()：检查所有服务器

agentring.mcp.EpisodeResults(results)

事件结果收集和分析。

方法：

• summary()：获取全面的统计数据
• to_json()：导出为JSON
• to_csv()：导出到CSV
• filter_by_success(successful_only)：筛选结果
• filter_by_reward(min_reward, max_reward)：按奖励筛选
• filter_by_steps(min_steps, max_steps)：按步骤筛选

实用函数

格式转换器

• agentring.mcp.formats.to_json_schema(tool)：转换为JSON模式
• agentring.mcp.formats.to_openapi_spec(tool)：转换为OpenAPI
• agentring.mcp.formats.to_crewai_tool(tool)：转换为CrewAI
• agentring.mcp.formats.to_langchain_tool(tool)：转换为LangChain

工具实用程序

• agentring.mcp.utils.compose_tools(*tool_lists)：合并工具列表
• agentring.mcp.utils.filter_tools(tools, names, include_patterns, exclude_patterns)：筛选工具
• agentring.mcp.utils.validate_tool_call(tool, args)：验证工具参数

模板

• agentring.mcp.templates.TEXT_ADVENTURE_INSTRUCTIONS：TextWorld说明
• agentring.mcp.templates.SHOPPING_INSTRUCTIONS：WebShop说明
• agentring.mcp.templates.HOUSEHOLD_INSTRUCTIONS：ALFWorld说明
• agentring.mcp.templates.GENERIC_INSTRUCTIONS：通用健身房说明

故障排除

常见问题

连接错误

# Check server health
client = gym_mcp.MCPServerClient("http://localhost:8070")
if not client.health_check():
    print("Server is not responding")

工具发现失败

# Try with explicit client
client = gym_mcp.MCPServerClient("http://localhost:8070")
try:
    tools = gym_mcp.create_tools("http://localhost:8070", client=client)
except Exception as e:
    print(f"Tool discovery failed: {e}")

SDK集成问题

# Validate tools before use
from agentring.mcp.utils import validate_tool_call

for tool in tools:
    is_valid, error = validate_tool_call(tool, {})
    if not is_valid:
        print(f"Tool {tool.__name__} has issues: {error}")

调试日志记录

import logging
logging.basicConfig(level=logging.DEBUG)

# This will show detailed MCP communication
tools = gym_mcp.create_tools("http://localhost:8070")

性能提示

1. 重复使用客户端：创建MCPServerClient一次并重复使用
2. 批量操作：使用 run_episodes() 而不是个人 run_episode() 电话
3. 缓存：服务器信息和工具会自动缓存
4. 健康检查：使用 health_check_all() 用于多服务器设置

相关项目

• 健身房mcp服务器 -体育馆环境下的MCP服务器
• 体育馆 -RL环境标准
• 模型上下文协议 -工具集成协议

支持

• 问题：打开GitHub问题
• 问题：开始GitHub讨论
• 文档：参见示例/目录

______________________________________________________________________

状态: ✅ 生产就绪| 版本: 0.4.0 | python: 3.10+ | 测试：19个核心+7个MCP扩展测试套件