File size: 4,163 Bytes
beb8990 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 | # Copyright (c) Meta Platforms, Inc. and affiliates.
# All rights reserved.
#
# This source code is licensed under the BSD-style license found in the
# LICENSE file in the root directory of this source tree.
"""
Generic Data Models for MCP Environment with Tool Support.
Following RFC 003 - Traditional Tool Calling Approach:
- MCPAction: Wrapper action that dispatches to ListToolsAction or ToolCallAction
- ListToolsAction: Discover available MCP tools
- ToolCallAction: Execute a specific MCP tool
These models are fully generic and work with any MCP integration.
"""
from typing import Any, Dict, List, Optional, Literal
from pydantic import Field
from openenv.core.env_server.types import Action, Observation
class MCPAction(Action):
"""
Generic wrapper action for MCP environment that supports multiple action types.
This action uses an action_type discriminator to determine which action to execute:
- "ListToolsAction": Discover available MCP tools
- "ToolCallAction": Execute a specific MCP tool
Args:
action_type: Type of action ("ListToolsAction" or "ToolCallAction")
tool_name: Name of tool to call (required for ToolCallAction)
arguments: Arguments for tool (optional, for ToolCallAction)
Examples:
>>> # List tools
>>> action = MCPAction(action_type="ListToolsAction")
>>>
>>> # Call a tool
>>> action = MCPAction(
... action_type="ToolCallAction",
... tool_name="create_resource",
... arguments={"name": "New Resource", "type": "example"}
... )
"""
action_type: Literal["ListToolsAction", "ToolCallAction"] = Field(
..., description="Type of action to perform"
)
tool_name: Optional[str] = Field(
None, description="Name of tool to call (required for ToolCallAction)"
)
arguments: Optional[Dict[str, Any]] = Field(
default_factory=dict, description="Arguments for the tool"
)
# Internal action types (not exposed to HTTP API)
class ListToolsAction(Action):
"""
Internal: Request list of available tools from MCP server.
This action corresponds to the MCP tools/list API.
Use MCPAction with action_type="ListToolsAction" instead.
"""
pass # No parameters needed
class ToolCallAction(Action):
"""
Internal: Call a specific MCP tool with arguments.
This action corresponds to the MCP tools/call API.
Use MCPAction with action_type="ToolCallAction" instead.
"""
tool_name: str = Field(..., description="Name of the tool to call")
arguments: Dict[str, Any] = Field(
default_factory=dict, description="Arguments to pass to the tool"
)
class MCPObservation(Observation):
"""
Generic observation from the MCP Environment.
Depending on the action type, different fields will be populated:
- For ListToolsAction: tools_list contains available tool schemas
- For ToolCallAction: tool_result contains the execution result
Args:
success: Whether the action succeeded
error_message: Error message if action failed
tools_list: List of available tools (for ListToolsAction)
tool_result: Result from tool execution (for ToolCallAction)
metadata: Additional metadata about the execution
done: Whether the episode is complete
reward: Reward for the action
"""
success: bool = Field(True, description="Whether the action succeeded")
error_message: Optional[str] = Field(None, description="Error message if action failed")
tools_list: Optional[List[Dict[str, Any]]] = Field(
None, description="List of available tools (for ListToolsAction)"
)
tool_result: Optional[Dict[str, Any]] = Field(
None, description="Result from tool execution (for ToolCallAction)"
)
metadata: Dict[str, Any] = Field(
default_factory=dict, description="Additional metadata about the execution"
)
done: bool = Field(False, description="Whether the episode is complete")
reward: Optional[float] = Field(None, description="Reward for the action") |