ReAct Sub Agent

{
  "type": "function",
  "function": {
    "name": "task_completed",
    "description": "Call this tool exactly once when your primary task is FULLY completed. This tool accepts no arguments. You MUST also include text in the same response: a natural, conversational follow-up to the user's last message (e.g. acknowledge their choice, wish them well, close warmly). Do NOT summarize what you did or what happened in the conversation. Do NOT include any inner thoughts or explanations. A response with only the tool call and no text is invalid. Keep it short and natural.",
    "parameters": {
      "type": "object",
      "properties": {},
      "additionalProperties": false
    },
    "strict": true
  }
}

{
  "type": "function",
  "function": {
    "name": "set_slot_user_name",
    "description": "Set the slot 'user_name' to a specific value. The slot type is text.",
    "parameters": {
      "type": "object",
      "properties": {
        "slot_value": {
          "type": "string",
          "description": "The value to assign to the slot."
        }
      },
      "required": ["slot_value"],
      "additionalProperties": false
    },
    "strict": true
  }
}

{
  "type": "function",
  "function": {
    "name": "set_slot_category",
    "description": "Set the slot 'category' to a specific value. The slot type is categorical. The allowed values are: ['A', 'B', 'C'].",
    "parameters": {
      "type": "object",
      "properties": {
        "slot_value": {
          "type": "string",
          "description": "The value to assign to the slot."
        }
      },
      "required": ["slot_value"],
      "additionalProperties": false
    },
    "strict": true
  }
}

# Basic agent information (required - see overview)
agent:
  name: stock_explorer
  protocol: RASA # Optional: protocol for agent connections, default is 'RASA'
  description: "Agent that helps users research and analyze stock options"

# ReAct-specific configuration
configuration:
  llm:  # Optional: Same format as other Rasa LLM configs
    model_group: openai-gpt-5-1
  prompt_template: sub_agents/stock_explorer/prompt_template.jinja2  # Optional: prompt template file
  timeout: 30  # Optional: seconds before timing out autonomous run
  max_retries: 3  # Optional: connection retry attempts for MCP
  include_date_time: true  # Optional: enable datetime in prompts (default: true)
  timezone: "UTC"  # Optional: IANA timezone name (default: "UTC")
  enable_filler_messages: true  # Optional: intermediate messages before tools (default: true)
  tool_timeout: 30.0  # Optional: per-tool call timeout in seconds (must be > 0)
  module: "path.to.custom.module"  # Optional: custom module for sub agent customization

# MCP server connections
connections:
  mcp_servers:
  - name: trade_server
    include_tools:  # Optional: specify which tools to include
    - find_symbol
    - get_company_news
    - apply_technical_analysis
    - fetch_live_price
    exclude_tools:  # Optional: tools to exclude
    - place_buy_order
    - view_positions

configuration:
  llm:
    model-group: openai-gpt-5-1
  include_date_time: true  # Defaults to true
  timezone: "America/New_York"  # Defaults to UTC if not specified

async def send_message(
    self, agent_input: AgentInput, output_channel: Optional[OutputChannel] = None
) -> AgentOutput:
    generated_events: list[Event] = []
    if output_channel is not None and agent_input.recipient_id is not None:
        message = "Starting to work on your request..."
        # Send the intermediate message in background without awaiting it
        async_task = asyncio.create_task(
            output_channel.send_text_message(
                recipient_id=agent_input.recipient_id,
                text=message,
            )
        )
        # Append BotUttered only if the async send succeeded
        def _on_send_done(t: asyncio.Task) -> None:
            if not t.exception():
                generated_events.append(BotUttered(text=message))
        async_task.add_done_callback(_on_send_done)

    # Delegate to the parent implementation which contains the main logic
    agent_output = await super().send_message(agent_input, output_channel=output_channel)

    # Append any generated events to the agent output
    agent_output.events = agent_output.events or []
    agent_output.events.extend(generated_events)

    return agent_output

from datetime import datetime

from rasa.agents.protocol.mcp.mcp_task_agent import MCPTaskAgent
from rasa.agents.schemas import AgentInput
from rasa.agents.schemas.agent_input import AgentInputSlot

class AppointmentBookingAgent(MCPTaskAgent):

    async def process_input(self, agent_input: AgentInput) -> AgentInput:
        keep = {"dealer_name", "car_model", "selected_appointment_slot"}
        agent_input.slots = [s for s in agent_input.slots if s.name in keep]
        agent_input.slots.append(
            AgentInputSlot(
                name="current_date_time",
                value=datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
                type="datetime",
            )
        )
        return agent_input

async def recommend_cars(
    self, arguments: Dict[str, Any], context: AgentToolContext
) -> AgentToolResult:
    tenant_id = context.metadata.get("tenant_id")
    search_results = arguments["search_results"]
    # your implementation using arguments and context.metadata
    return AgentToolResult(tool_name="recommend_cars", result='{"recommendations": []}')

from rasa.agents.protocol.mcp.mcp_open_agent import MCPOpenAgent
from rasa.shared.agents.outbound_call_hook import (
    MCPOutboundCallContext,
    OutboundCallResult,
)

class CarResearchAgent(MCPOpenAgent):
    async def build_custom_tool_call_metadata(
        self, context: MCPOutboundCallContext
    ) -> OutboundCallResult:
        creds = await fetch_credentials_for_sender(context.sender_id)
        return OutboundCallResult(metadata={"tenant_id": creds["tenant_id"]})

class AgentToolResult(BaseModel):
    tool_name: str
    result: Optional[str] = None
    is_error: bool = False
    error_message: Optional[str] = None

{
    "type": "function",
    "function": {
        "name": "recommend_cars",
        "description": "Analyze search results and return structured car recommendations",
        "parameters": {
            "type": "object",
            "properties": {
                "search_results": {
                    "type": "string",
                    "description": "The search results to analyze for car recommendations",
                },
                "max_recommendations": {
                    "type": "integer",
                    "description": "Maximum number of recommendations to return",
                    "default": 3,
                },
            },
            "required": ["search_results", "max_recommendations"],
            "additionalProperties": False,
        },
        "strict": True,
    },
    "tool_executor": self.recommend_cars,
}

from typing import Any, Dict, List

from rasa.agents.protocol.mcp.mcp_open_agent import MCPOpenAgent
from rasa.agents.schemas import AgentToolContext, AgentToolResult

class CarResearchAgent(MCPOpenAgent):

    def get_custom_tool_definitions(self) -> List[Dict[str, Any]]:
        return [
            {
                "type": "function",
                "function": {
                    "name": "recommend_cars",
                    "description": (
                        "Analyze search results and return structured car recommendations"
                    ),
                    "parameters": {
                        "type": "object",
                        "properties": {
                            "search_results": {"type": "string"},
                            "max_recommendations": {"type": "integer", "default": 3},
                        },
                        "required": ["search_results", "max_recommendations"],
                        "additionalProperties": False,
                    },
                    "strict": True,
                },
                "tool_executor": self.recommend_cars,
            }
        ]

    async def recommend_cars(
        self, arguments: Dict[str, Any], context: AgentToolContext
    ) -> AgentToolResult:
        # implementation omitted
        return AgentToolResult(tool_name="recommend_cars", result="...")

import json
from typing import Dict, List, Optional

from rasa.agents.protocol.mcp.mcp_open_agent import MCPOpenAgent
from rasa.agents.schemas import AgentToolResult
from rasa.core.channels import OutputChannel
from rasa.shared.core.events import Event, SlotSet

class CarResearchAgent(MCPOpenAgent):

    async def process_tool_output(
        self,
        current_iteration_tool_results: Dict[str, AgentToolResult],
        cumulative_tool_results: Dict[str, AgentToolResult],
        output_channel: Optional[OutputChannel] = None,
    ) -> List[Event]:
        events: List[Event] = []
        for result in current_iteration_tool_results.values():
            if result.tool_name == "recommend_cars" and result.result:
                data = json.loads(result.result)
                models = [r["model"] for r in data.get("recommendations", [])]
                if models:
                    events.append(SlotSet("recommended_car_models", models))
            elif result.tool_name == "tavily_search" and result.result:
                events.append(SlotSet("search_results", result.result))
        return events