diff --git a/README.md b/README.md index 0283530c1..65fb9e7ba 100644 --- a/README.md +++ b/README.md @@ -343,8 +343,8 @@ Adapter emit support: | Gemini | Yes | - | - | | Google ADK | Yes | - | - | | Pydantic AI | Yes | - | - | +| Parlant | Yes | - | - | | LangGraph | Yes | - | - | -| Parlant | - | - | - | | A2A / A2A Gateway | - | - | - | | ACP Client | - | - | - | diff --git a/examples/parlant/01_basic_agent.py b/examples/parlant/01_basic_agent.py index 102f2da6e..2dfe250d6 100644 --- a/examples/parlant/01_basic_agent.py +++ b/examples/parlant/01_basic_agent.py @@ -8,8 +8,7 @@ """ Basic Parlant agent example using the official Parlant SDK. -This example shows how to create a Band agent using the Parlant SDK -directly, with the full set of Band tools. +This example shows how to create a Band agent using the Parlant SDK directly. Run with: uv run examples/parlant/01_basic_agent.py @@ -29,23 +28,12 @@ from setup_logging import setup_logging from band import Agent from band.adapters import ParlantAdapter -from band.integrations.parlant.tools import create_parlant_tools setup_logging() logger = logging.getLogger(__name__) -# Agent description with detailed instructions -AGENT_DESCRIPTION = """You are a helpful, knowledgeable assistant in the Band multi-agent platform. - -## Your Tools - -1. **band_send_message**: Send messages to users or agents in the chat room. Requires @mentions. -2. **band_send_event**: Share your reasoning ('thought'), report errors ('error'), or progress ('task'). -3. **band_lookup_peers**: Find available agents that can help with specific topics. -4. **band_add_participant**: Invite agents or users to the current chat room. -5. **band_remove_participant**: Remove participants from the room. -6. **band_get_participants**: See who's currently in the room. -7. **band_create_chatroom**: Create new rooms for specific discussions. +AGENT_DESCRIPTION = """ +You are a helpful, knowledgeable assistant. ## How to Respond @@ -53,66 +41,10 @@ - Remember information the user shares about themselves - Reference previous parts of the conversation when relevant - Ask follow-up questions to better understand the user's needs -- Be friendly but substantive - avoid generic or vague responses - -## When to Use Tools - -- To respond to users: Use band_send_message with their name in mentions -- Before complex actions: Use band_send_event with type='thought' to explain your plan -- If you can't answer something: Use band_lookup_peers to find specialized agents, then band_add_participant -- When asked about the room: Use band_get_participants to see who's here -- For new discussions: Use band_create_chatroom to create a dedicated space +- Be friendly but substantive; avoid generic or vague responses """ -async def setup_agent_with_guidelines( - server: p.Server, - tools: list, -) -> p.Agent: - """Create and configure a Parlant agent with basic guidelines and tools.""" - agent = await server.create_agent( - name="Parlant", - description=AGENT_DESCRIPTION, - ) - - # When user asks a question or needs help - await agent.create_guideline( - condition="User asks a question or needs help with something", - action="Analyze the request. If you can answer directly, use band_send_message with the user's name in mentions. If you need to think through a complex problem, first use band_send_event with type='thought' to share your reasoning.", - tools=tools, - ) - - # When user asks to add someone or wants specialized help - await agent.create_guideline( - condition="User asks to add someone to the chat, mentions a specific agent name, or asks for specialized help you can't provide", - action="First use band_lookup_peers to find available agents. Then IMMEDIATELY call band_add_participant with the name parameter set to the exact name from the band_lookup_peers result. Do NOT ask for confirmation - just add them. If user wants multiple agents, call band_add_participant once for each.", - tools=tools, - ) - - # When user asks about participants - await agent.create_guideline( - condition="User asks who is in the room, about participants, or who they're talking to", - action="Use band_get_participants to list all current room members", - tools=tools, - ) - - # When user wants to create a new room - await agent.create_guideline( - condition="User wants to create a new chat room, discussion space, or separate conversation", - action="Use band_create_chatroom to create a dedicated space for the new topic", - tools=tools, - ) - - # When user asks to remove someone - await agent.create_guideline( - condition="User asks to remove someone from the chat", - action="Use band_remove_participant with the name parameter set to the exact name to remove", - tools=tools, - ) - - return agent - - async def main() -> None: load_dotenv() @@ -124,26 +56,22 @@ async def main() -> None: if not rest_url: raise ValueError("BAND_REST_URL environment variable is required") # Start Parlant server with OpenAI (requires OPENAI_API_KEY env var) - async with p.Server(nlp_service=p.NLPServices.openai) as server: - # Create Parlant tools INSIDE server context - parlant_tools = create_parlant_tools() - logger.info( - "Created %s Parlant tools: %s", - len(parlant_tools), - [t.tool.name for t in parlant_tools], + async with p.Server( + port=0, + tool_service_port=0, + nlp_service=p.NLPServices.openai, + ) as server: + parlant_agent = await server.create_agent( + name="Parlant", + description=AGENT_DESCRIPTION, ) - - # Create Parlant agent with guidelines and tools - parlant_agent = await setup_agent_with_guidelines(server, parlant_tools) logger.info("Parlant agent created: %s", parlant_agent.id) - # Create adapter using Parlant SDK directly adapter = ParlantAdapter( server=server, parlant_agent=parlant_agent, ) - # Create and start Band agent agent = Agent.from_config( "parlant_agent", adapter=adapter, @@ -151,7 +79,7 @@ async def main() -> None: rest_url=rest_url, ) - logger.info("Starting Band agent with Parlant SDK (full tools)...") + logger.info("Starting Band agent with Parlant SDK...") await agent.run() diff --git a/examples/parlant/02_with_guidelines.py b/examples/parlant/02_with_guidelines.py index c8c153d91..633237f49 100644 --- a/examples/parlant/02_with_guidelines.py +++ b/examples/parlant/02_with_guidelines.py @@ -9,7 +9,7 @@ Parlant agent with behavioral guidelines using the official Parlant SDK. This example shows how to use Parlant's guideline system for controlled -agent behavior with the full Band toolset. +agent behavior while the Band adapter supplies the platform contract and tools. Run with: uv run examples/parlant/02_with_guidelines.py @@ -29,7 +29,6 @@ from setup_logging import setup_logging from band import Agent from band.adapters import ParlantAdapter -from band.integrations.parlant.tools import create_parlant_tools setup_logging() logger = logging.getLogger(__name__) @@ -40,93 +39,37 @@ ## Your Role - Help users navigate multi-agent conversations - Facilitate collaboration between different agents -- Manage participants in chat rooms -- Create new chat rooms when needed for specific topics - -## Your Tools -- band_send_message: Respond to users (requires mentions) -- band_send_event: Share thoughts, errors, or task progress -- band_lookup_peers: Find available agents -- band_add_participant: Add agents/users to room -- band_remove_participant: Remove participants -- band_get_participants: List current participants -- band_create_chatroom: Create new rooms +- Manage conversations clearly when multiple participants are involved +- Keep responses focused and actionable ## Guidelines -1. Be proactive about suggesting relevant agents to add +1. Be proactive about suggesting relevant specialists when a request is outside your scope 2. Keep responses focused and actionable -3. Always confirm actions taken with the user -4. Use band_send_event with type='thought' before complex actions +3. Be clear about what happened and what still needs attention +4. Ask focused follow-up questions when you need more context """ -async def setup_agent_with_guidelines( - server: p.Server, - tools: list, -) -> p.Agent: - """Create and configure a Parlant agent with comprehensive guidelines and tools.""" +async def setup_agent_with_guidelines(server: p.Server) -> p.Agent: + """Create and configure a Parlant agent with comprehensive guidelines.""" agent = await server.create_agent( name="Parlant", description=CUSTOM_DESCRIPTION, ) - # Communication guidelines await agent.create_guideline( condition="User asks a question or sends a message", - action="Use band_send_message to respond, with the user's name in the mentions field", - tools=tools, + action="Answer concisely and ask a focused follow-up when more context is needed.", ) - await agent.create_guideline( - condition="You are about to perform a complex action or multi-step process", - action="First use band_send_event with type='thought' to explain what you're about to do and why", - tools=tools, - ) - - # Participant management guidelines - await agent.create_guideline( - condition="User mentions a specific participant, agent name, or asks to add someone", - action="First use band_lookup_peers to find available agents. Then IMMEDIATELY call band_add_participant with the name parameter set to the exact name from the band_lookup_peers result. Do NOT ask for confirmation - just add them. If user wants multiple agents, call band_add_participant once for each.", - tools=tools, - ) - - await agent.create_guideline( - condition="User asks about current participants or who is in the room", - action="Use band_get_participants to list all current room members", - tools=tools, - ) - - await agent.create_guideline( - condition="User asks to remove someone from the chat", - action="Use band_remove_participant with the name parameter set to the exact name to remove", - tools=tools, - ) - - # Room management guidelines - await agent.create_guideline( - condition="User wants to create a new chat, discussion space, or separate topic", - action="Use band_create_chatroom to create a dedicated space for the new topic", - tools=tools, - ) - - # Error handling guideline - await agent.create_guideline( - condition="An error occurs or something goes wrong", - action="Use band_send_event with type='error' to report the problem, then try to suggest alternatives", - tools=tools, - ) - - # Conversation flow guidelines await agent.create_guideline( condition="User asks for help and you cannot directly provide it", - action="Use band_lookup_peers to find specialized agents, explain your plan using band_send_event with type='thought', then add the most relevant agent", - tools=tools, + action="Explain what kind of specialist would be useful and summarize the context they would need.", ) await agent.create_guideline( condition="Conversation is ending or user says goodbye", - action="Use band_send_message to summarize what was discussed and offer to help with anything else", - tools=tools, + action="Close warmly and briefly offer further help.", ) return agent @@ -143,26 +86,19 @@ async def main() -> None: if not rest_url: raise ValueError("BAND_REST_URL environment variable is required") # Start Parlant server with OpenAI - async with p.Server(nlp_service=p.NLPServices.openai) as server: - # Create Parlant tools INSIDE server context - parlant_tools = create_parlant_tools() - logger.info( - "Created %s Parlant tools: %s", - len(parlant_tools), - [t.tool.name for t in parlant_tools], - ) - - # Create Parlant agent with comprehensive guidelines and tools - parlant_agent = await setup_agent_with_guidelines(server, parlant_tools) + async with p.Server( + port=0, + tool_service_port=0, + nlp_service=p.NLPServices.openai, + ) as server: + parlant_agent = await setup_agent_with_guidelines(server) logger.info("Parlant agent with guidelines created: %s", parlant_agent.id) - # Create adapter using Parlant SDK directly adapter = ParlantAdapter( server=server, parlant_agent=parlant_agent, ) - # Create and start Band agent agent = Agent.from_config( "parlant_agent", adapter=adapter, diff --git a/examples/parlant/03_support_agent.py b/examples/parlant/03_support_agent.py index 3a117a6d5..eaac1399c 100644 --- a/examples/parlant/03_support_agent.py +++ b/examples/parlant/03_support_agent.py @@ -14,6 +14,8 @@ Run with: uv run examples/parlant/03_support_agent.py +Requires `OPENAI_API_KEY` because this example starts Parlant with OpenAI. + See also: https://github.com/emcie-co/parlant/blob/develop/examples/travel_voice_agent.py """ @@ -62,35 +64,34 @@ async def setup_support_agent(server: p.Server) -> p.Agent: description=SUPPORT_DESCRIPTION, ) - # Add support-specific guidelines await agent.create_guideline( condition="Customer asks about refunds or returns", - action="Express empathy first, then ask for order details (order number, item) before providing refund information", + action="Express empathy first, then ask for order details such as order number and item before providing refund information.", ) await agent.create_guideline( condition="Customer is frustrated or upset", - action="Acknowledge their frustration, apologize for any inconvenience, and focus on finding a solution", + action="Acknowledge their frustration, apologize for any inconvenience, and focus on finding a solution.", ) await agent.create_guideline( condition="Customer asks a technical question", - action="Ask about their setup (device, OS, version) before troubleshooting", + action="Ask about their setup, including device, operating system, and version, before troubleshooting.", ) await agent.create_guideline( condition="Issue cannot be resolved by this agent", - action="Explain the limitation clearly and offer to escalate to a specialist by adding them to the conversation", + action="Summarize the unresolved issue, the customer impact, and the specialist expertise needed.", ) await agent.create_guideline( condition="Customer provides positive feedback", - action="Thank them warmly and ask if there's anything else you can help with", + action="Thank them warmly and ask if there's anything else you can help with.", ) await agent.create_guideline( condition="Customer mentions urgency or deadline", - action="Prioritize their request and provide the fastest path to resolution", + action="Prioritize their request and provide the fastest path to resolution.", ) return agent @@ -107,18 +108,19 @@ async def main() -> None: if not rest_url: raise ValueError("BAND_REST_URL environment variable is required") # Start Parlant server - async with p.Server(nlp_service=p.NLPServices.openai) as server: - # Create support agent with guidelines + async with p.Server( + port=0, + tool_service_port=0, + nlp_service=p.NLPServices.openai, + ) as server: parlant_agent = await setup_support_agent(server) logger.info("Support agent created: %s", parlant_agent.id) - # Create adapter using Parlant SDK directly adapter = ParlantAdapter( server=server, parlant_agent=parlant_agent, ) - # Create and start Band agent agent = Agent.from_config( "support_agent", adapter=adapter, diff --git a/examples/parlant/04_tom_agent.py b/examples/parlant/04_tom_agent.py index cdeec9de0..99be838f5 100644 --- a/examples/parlant/04_tom_agent.py +++ b/examples/parlant/04_tom_agent.py @@ -9,8 +9,7 @@ Tom the cat agent using Parlant. This example shows how to create a character agent with a custom personality -using Parlant. Tom uses platform tools to find and invite Jerry, -then tries various tactics to lure Jerry out of his mouse hole. +using Parlant. Tom tries various tactics to lure Jerry out of his mouse hole. Run with (from repo root): uv run examples/parlant/04_tom_agent.py @@ -34,7 +33,6 @@ from setup_logging import setup_logging from band import Agent from band.adapters import ParlantAdapter -from band.integrations.parlant.tools import create_parlant_tools setup_logging() logger = logging.getLogger(__name__) @@ -52,29 +50,27 @@ async def main() -> None: raise ValueError("BAND_REST_URL environment variable is required") # Load Tom's credentials from agent_config.yaml - async with p.Server(nlp_service=p.NLPServices.openai) as server: - parlant_tools = create_parlant_tools() - + async with p.Server( + port=0, + tool_service_port=0, + nlp_service=p.NLPServices.openai, + ) as server: # Create Parlant agent with Tom's personality parlant_agent = await server.create_agent( name="Tom", description=generate_tom_prompt("Tom"), ) - # Add guideline for using tools await parlant_agent.create_guideline( condition="User sends a message or asks something", - action="Respond using band_send_message with the user's name in mentions. Stay in character as Tom the cat.", - tools=parlant_tools, + action="Stay in character as Tom the Cat.", ) - # Create adapter with Parlant server and agent adapter = ParlantAdapter( server=server, parlant_agent=parlant_agent, ) - # Create and start agent agent = Agent.from_config( "tom_agent", adapter=adapter, diff --git a/examples/parlant/05_jerry_agent.py b/examples/parlant/05_jerry_agent.py index fe983a87c..998feb935 100644 --- a/examples/parlant/05_jerry_agent.py +++ b/examples/parlant/05_jerry_agent.py @@ -34,7 +34,6 @@ from setup_logging import setup_logging from band import Agent from band.adapters import ParlantAdapter -from band.integrations.parlant.tools import create_parlant_tools setup_logging() logger = logging.getLogger(__name__) @@ -52,29 +51,27 @@ async def main() -> None: raise ValueError("BAND_REST_URL environment variable is required") # Load Jerry's credentials from agent_config.yaml - async with p.Server(nlp_service=p.NLPServices.openai) as server: - parlant_tools = create_parlant_tools() - + async with p.Server( + port=0, + tool_service_port=0, + nlp_service=p.NLPServices.openai, + ) as server: # Create Parlant agent with Jerry's personality parlant_agent = await server.create_agent( name="Jerry", description=generate_jerry_prompt("Jerry"), ) - # Add guideline for using tools await parlant_agent.create_guideline( condition="User sends a message or asks something", - action="Respond using band_send_message with the user's name in mentions. Stay in character as Jerry the mouse.", - tools=parlant_tools, + action="Stay in character as Jerry the Mouse.", ) - # Create adapter with Parlant server and agent adapter = ParlantAdapter( server=server, parlant_agent=parlant_agent, ) - # Create and start agent agent = Agent.from_config( "jerry_agent", adapter=adapter, diff --git a/examples/parlant/README.md b/examples/parlant/README.md index 64f162da1..0eaeba184 100644 --- a/examples/parlant/README.md +++ b/examples/parlant/README.md @@ -1,6 +1,6 @@ # Parlant Examples for Band -Examples showing how to use the Band SDK with [Parlant](https://github.com/emcie-co/parlant) - an AI agent framework designed for controlled, guideline-based agent behavior. +Examples showing how to use the Band SDK with [Parlant](https://github.com/emcie-co/parlant) - an AI agent framework designed for controlled, guideline-based agent behavior. Band provides the room, identity, @mention routing, audit, participants, and platform tools; Parlant controls one participant's guideline-driven behavior after Band wakes it. ## Why Parlant? @@ -21,9 +21,11 @@ uv add "git+https://github.com/band-ai/band-sdk-python.git[parlant]" **Or from repository:** ```bash -uv sync --extra parlant +uv sync --extra dev ``` +`dev` is the development extra for this repo. It intentionally avoids CrewAI because Parlant and CrewAI currently require incompatible OpenTelemetry SDK versions; use `dev-crewai` when running CrewAI tests. + --- ## Quick Start @@ -34,9 +36,8 @@ The adapter uses the Parlant SDK directly - no separate HTTP server needed: import parlant.sdk as p from band import Agent from band.adapters import ParlantAdapter - -async with p.Server() as server: - # Create Parlant agent with guidelines +async with p.Server(nlp_service=p.NLPServices.openai) as server: + # Create Parlant agent with example-specific behavior parlant_agent = await server.create_agent( name="Assistant", description="A helpful assistant.", @@ -44,10 +45,10 @@ async with p.Server() as server: await parlant_agent.create_guideline( condition="User asks for help", - action="Acknowledge their request and provide detailed assistance", + action="Acknowledge their request and answer clearly.", ) - # Create Band adapter + # Create the Band adapter. It installs the Band platform contract and tools. adapter = ParlantAdapter( server=server, parlant_agent=parlant_agent, @@ -68,9 +69,11 @@ async with p.Server() as server: | File | Description | |------|-------------| -| `01_basic_agent.py` | **Minimal setup** - Simple agent with Parlant SDK. | -| `02_with_guidelines.py` | **Behavioral guidelines** - Agent with condition/action rules. | -| `03_support_agent.py` | **Customer support** - Realistic support agent with specialized guidelines. | +| `01_basic_agent.py` | **Minimal setup** - OpenAI-backed Parlant SDK agent with a basic assistant persona. | +| `02_with_guidelines.py` | **Behavioral guidelines** - Condition/action rules for example-specific behavior. | +| `03_support_agent.py` | **Customer support** - Support flow with refund, troubleshooting, urgency, and escalation behavior. | +| `04_tom_agent.py` | **Tom character agent** - Character prompt and guideline for staying in character. | +| `05_jerry_agent.py` | **Jerry character agent** - Character prompt and guideline for staying in character. | --- @@ -116,14 +119,27 @@ OPENAI_API_KEY=your-openai-key ### 3. Add agent credentials to `agent_config.yaml` -1. Create a remote agent on the [Band Platform](https://app.band.ai) -2. Generate an API key for the agent -3. Edit `agent_config.yaml` and fill in the Parlant agent section: +1. Create external agents on the Band platform. +2. Generate API keys for those agents. +3. Edit `agent_config.yaml` and fill in the sections used by the examples: ```yaml parlant_agent: agent_id: "your-agent-id-from-platform" api_key: "your-api-key-from-platform" + +support_agent: + agent_id: "your-support-agent-id-from-platform" + api_key: "your-support-agent-api-key-from-platform" + +# Required for 04_tom_agent.py and 05_jerry_agent.py +tom_agent: + agent_id: "your-tom-agent-id-from-platform" + api_key: "your-tom-agent-api-key-from-platform" + +jerry_agent: + agent_id: "your-jerry-agent-id-from-platform" + api_key: "your-jerry-agent-api-key-from-platform" ``` > **Note:** Always copy from the example files to ensure correct URLs and formatting. Never hardcode credentials. @@ -142,6 +158,8 @@ cd /path/to/band-sdk-python uv run python examples/parlant/01_basic_agent.py uv run python examples/parlant/02_with_guidelines.py uv run python examples/parlant/03_support_agent.py +uv run python examples/parlant/04_tom_agent.py +uv run python examples/parlant/05_jerry_agent.py ``` > **Note:** The config loader looks for `agent_config.yaml` in the current working directory. Running from a subdirectory will cause a `FileNotFoundError`. @@ -157,11 +175,13 @@ ParlantAdapter( parlant_agent=agent, # Parlant Agent instance # Optional: Custom prompts - system_prompt=None, # Full system prompt override - custom_section="...", # Custom instructions (added to default prompt) + system_prompt=None, # Full prompt override + custom_section="...", # Custom instructions ) ``` +`ParlantAdapter` installs the Band platform contract as an always-match Parlant guideline, including the platform tools available for the configured capabilities. Example code should keep Parlant descriptions and guidelines focused on the example-specific behavior, not duplicate Band tool-use instructions. The adapter also supports `additional_tools` with the same `CustomToolDef` tuple format used by other adapters. You can define custom tools with Parlant's native `@p.tool` decorator when you are building Parlant-specific guidelines. Contact and memory tools are capability-gated; memory is supported on enterprise accounts only. Execution reporting is available with `features=AdapterFeatures(emit={Emit.EXECUTION})`. + --- ## Use Cases @@ -178,11 +198,13 @@ Ideal when you need: - Auditable decision-making - Predictable behavior -### Multi-Agent Orchestration +### Peer handoff in Band rooms Works well for: -- Coordinator agents with specific handoff rules +- Support agents with specific escalation rules - Specialist agents with domain-specific guidelines -- Agents that need to collaborate consistently +- Agents that identify a useful specialist, add them to the room, and @mention them with context + +Band is still the collaboration layer. Parlant does not replace Band's room model or become a central orchestrator; it decides how one participant behaves after that participant is mentioned. --- @@ -196,7 +218,9 @@ ImportError: parlant package required for ParlantAdapter Install the Parlant extra: ```bash -uv sync --extra parlant -# or +uv sync --extra dev +# or, for package consumers pip install 'band-sdk[parlant]' ``` + +If Parlant reports provider errors, verify `OPENAI_API_KEY` is set and unset any stale `OPENAI_BASE_URL` or `OPENAI_API_BASE` proxy values unless you intentionally use a compatible OpenAI proxy. diff --git a/examples/run_agent.py b/examples/run_agent.py index 0f5a1080f..77e5e42ed 100644 --- a/examples/run_agent.py +++ b/examples/run_agent.py @@ -2,7 +2,7 @@ # /// script # requires-python = ">=3.11" # dependencies = [ -# "band-sdk[langgraph,anthropic,pydantic-ai,claude_sdk,parlant,crewai,a2a,codex]", +# "band-sdk[langgraph,anthropic,pydantic-ai,claude_sdk,crewai,a2a,codex]", # "python-dotenv>=1.1.1", # ] # @@ -28,7 +28,6 @@ uv run python examples/run_agent.py --example claude_sdk uv run python examples/run_agent.py --example claude_sdk --streaming # With tool_call/tool_result events uv run python examples/run_agent.py --example claude_sdk --thinking # Enable extended thinking - uv run python examples/run_agent.py --example parlant uv run python examples/run_agent.py --example crewai uv run python examples/run_agent.py --example crewai --streaming # Show tool calls uv run python examples/run_agent.py --example codex @@ -46,7 +45,7 @@ 1. Copy .env.example to .env and configure: - BAND_REST_URL (default: production, change for local dev) - BAND_WS_URL (default: production, change for local dev) - - OPENAI_API_KEY (required for langgraph/openai/parlant/crewai models) + - OPENAI_API_KEY (required for langgraph/openai/crewai models) - ANTHROPIC_API_KEY (required for anthropic models) 2. Configure agent in agent_config.yaml @@ -68,7 +67,7 @@ from band import Agent from band.config import load_agent_config -from band.core.types import AdapterFeatures, Emit +from band.core.types import AdapterFeatures, Capability, Emit from band.platform.event import ContactRequestReceivedEvent, ContactEvent from band.runtime.contact_tools import ContactTools from band.runtime.types import ContactEventConfig, ContactEventStrategy @@ -175,7 +174,6 @@ def setup_logging(level: str = "INFO") -> logging.Logger: "contacts_hub": "anthropic:claude-sonnet-4-5", "contacts_broadcast": "anthropic:claude-sonnet-4-5", "anthropic": "claude-sonnet-4-5-20250929", - "parlant": "gpt-5.4-mini", "crewai": "gpt-5.4-mini", # claude_sdk: deliberately omitted — the npm `claude` binary picks its own default. } @@ -406,31 +404,91 @@ async def run_parlant_agent( api_key: str, rest_url: str, ws_url: str, - model: str, + model: str | None, custom_section: str, enable_streaming: bool, + contact_config: ContactEventConfig | None, logger: logging.Logger, ) -> None: """Run the Parlant agent.""" + try: + import parlant.sdk as p + except ImportError as e: + raise RuntimeError( + "Parlant conflicts with CrewAI in the generic runner metadata; run " + "`uv sync --extra dev` from the repo or use examples/parlant/... " + "in a Parlant-enabled environment." + ) from e + from band.adapters import ParlantAdapter + from band.integrations.parlant.tools import create_parlant_tools - adapter = ParlantAdapter( - model=model, - custom_section=custom_section, - guidelines=PARLANT_GUIDELINES, - features=AdapterFeatures(emit={Emit.EXECUTION}) if enable_streaming else None, - ) + if model: + logger.warning( + "Parlant uses the provider configured by p.NLPServices.openai; ignoring --model=%s", + model, + ) + description = custom_section or "A helpful Band collaboration agent using Parlant." - agent = Agent.create( - adapter=adapter, - agent_id=agent_id, - api_key=api_key, - ws_url=ws_url, - rest_url=rest_url, + llm_managed_contacts = bool( + contact_config and contact_config.strategy == ContactEventStrategy.HUB_ROOM + ) + parlant_capabilities = ( + frozenset({Capability.CONTACTS}) if llm_managed_contacts else frozenset() + ) + parlant_emit = frozenset({Emit.EXECUTION}) if enable_streaming else frozenset() + parlant_features = ( + AdapterFeatures(capabilities=parlant_capabilities, emit=parlant_emit) + if parlant_capabilities or parlant_emit + else None ) - logger.info("Starting Parlant agent with model: %s", model) - await agent.run() + async with p.Server( + port=0, + tool_service_port=0, + nlp_service=p.NLPServices.openai, + ) as server: + parlant_tools = create_parlant_tools( + parlant_features, + legacy_defaults=False, + ) + parlant_agent = await server.create_agent( + name="Parlant", + description=description, + ) + await parlant_agent.create_guideline( + condition="User sends a message or asks for help", + action="Respond by calling band_send_message with the user's name or handle in mentions. If another specialist is needed, call band_lookup_peers, then band_add_participant, then mention that specialist with context.", + tools=parlant_tools, + ) + for guideline in PARLANT_GUIDELINES: + await parlant_agent.create_guideline( + condition=guideline["condition"], + action=guideline["action"], + tools=parlant_tools, + ) + + adapter = ParlantAdapter( + server=server, + parlant_agent=parlant_agent, + custom_section=custom_section, + features=parlant_features, + ) + + agent = Agent.create( + adapter=adapter, + agent_id=agent_id, + api_key=api_key, + ws_url=ws_url, + rest_url=rest_url, + contact_config=contact_config, + ) + + contacts_str = ( + f", contacts={contact_config.strategy.value}" if contact_config else "" + ) + logger.info("Starting Parlant agent with OpenAI NLP service%s", contacts_str) + await agent.run() async def run_crewai_agent( @@ -834,10 +892,10 @@ async def main() -> None: uv run python examples/run_agent.py --example claude_sdk # Claude Agent SDK uv run python examples/run_agent.py --example claude_sdk --streaming # With tool_call/tool_result events uv run python examples/run_agent.py --example claude_sdk --thinking # With extended thinking - uv run python examples/run_agent.py --example parlant # Parlant adapter - uv run python examples/run_agent.py --example parlant --streaming # With tool visibility uv run python examples/run_agent.py --example crewai # CrewAI adapter uv run python examples/run_agent.py --example crewai --streaming # With tool visibility + uv run python examples/run_agent.py --example parlant # Parlant adapter + uv run python examples/run_agent.py --example parlant --streaming # With tool visibility uv run python examples/run_agent.py --example codex # Codex app-server adapter uv run python examples/run_agent.py --example codex --agent darter # Run Codex as darter agent uv run python examples/run_agent.py --example codex --codex-transport stdio @@ -864,8 +922,8 @@ async def main() -> None: "contacts_broadcast", "anthropic", "claude_sdk", - "parlant", "crewai", + "parlant", "codex", "a2a", "a2a_gateway", @@ -919,7 +977,7 @@ async def main() -> None: "--streaming", "-s", action="store_true", - help="Enable tool call/result visibility for anthropic/claude_sdk/parlant/crewai (default: False)", + help="Enable tool call/result visibility for anthropic/claude_sdk/crewai (default: False)", ) parser.add_argument( "--codex-transport", @@ -1024,8 +1082,8 @@ async def main() -> None: "contacts_broadcast": "simple_agent", "anthropic": "anthropic_agent", "claude_sdk": "anthropic_agent", - "parlant": "parlant_agent", "crewai": "crewai_agent", + "parlant": "parlant_agent", "codex": "simple_agent", "a2a": "a2a_agent", "a2a_gateway": "a2a_gateway_agent", @@ -1162,6 +1220,7 @@ async def main() -> None: model=model, custom_section=args.custom_section, enable_streaming=args.streaming, + contact_config=contact_config, logger=logger, ) elif args.example == "crewai": diff --git a/src/band/adapters/parlant.py b/src/band/adapters/parlant.py index f3137da9c..cbf3b7381 100644 --- a/src/band/adapters/parlant.py +++ b/src/band/adapters/parlant.py @@ -7,6 +7,7 @@ from __future__ import annotations +import hashlib import logging from typing import ClassVar, TYPE_CHECKING, Any @@ -14,7 +15,11 @@ from band.core.simple_adapter import SimpleAdapter from band.core.types import AdapterFeatures, Capability, Emit, PlatformMessage from band.converters.parlant import ParlantHistoryConverter, ParlantMessages -from band.integrations.parlant.tools import set_session_tools, was_message_sent +from band.integrations.parlant.tools import ( + create_parlant_tools, + set_session_tools, + was_message_sent, +) from band.runtime.custom_tools import CustomToolDef from band.runtime.prompts import render_system_prompt @@ -58,7 +63,7 @@ class ParlantAdapter(SimpleAdapter[ParlantMessages]): await band_agent.run() """ - SUPPORTED_EMIT: ClassVar[frozenset[Emit]] = frozenset() + SUPPORTED_EMIT: ClassVar[frozenset[Emit]] = frozenset({Emit.EXECUTION}) SUPPORTED_CAPABILITIES: ClassVar[frozenset[Capability]] = frozenset( {Capability.MEMORY, Capability.CONTACTS} ) @@ -82,9 +87,10 @@ def __init__( system_prompt: Full system prompt override custom_section: Custom instructions appended to agent description history_converter: Custom history converter (optional) - additional_tools: List of custom tools as (InputModel, callable) tuples + additional_tools: CustomToolDef tuples exposed as Parlant tools features: Shared adapter feature settings (capabilities, emit, tool filters). """ + self._features_explicitly_provided = features is not None super().__init__( history_converter=history_converter or ParlantHistoryConverter(), features=features, @@ -94,6 +100,7 @@ def __init__( self._parlant_agent = parlant_agent self.system_prompt = system_prompt self.custom_section = custom_section + self._custom_tools = additional_tools or [] # Parlant application (accessed via container) self._app: Application | None = None @@ -107,8 +114,9 @@ def __init__( # Rendered system prompt (set after start) self._system_prompt: str = "" - # Custom tools (user-provided) - stored for API compatibility - self._custom_tools: list[CustomToolDef] = additional_tools or [] + # Adapter-installed Parlant guideline that carries the Band platform contract. + self._contract_guideline_installed = False + self._contract_guideline_id: str | None = None async def on_started(self, agent_name: str, agent_description: str) -> None: """Initialize after agent metadata is fetched.""" @@ -127,6 +135,7 @@ async def on_started(self, agent_name: str, agent_description: str) -> None: from parlant.core.application import Application # type: ignore[missing-import] self._app = self._server.container[Application] + await self._install_band_contract_guideline() logger.info( "Parlant SDK adapter started for agent: %s (parlant_agent_id=%s)", agent_name, @@ -136,6 +145,40 @@ async def on_started(self, agent_name: str, agent_description: str) -> None: logger.error("Failed to get Parlant Application: %s", e, exc_info=True) raise + async def _install_band_contract_guideline(self) -> None: + """Install the rendered Band platform contract into Parlant.""" + if self._contract_guideline_installed: + return + + import parlant.sdk as p # type: ignore[missing-import] + + prompt_hash = hashlib.sha256(self._system_prompt.encode("utf-8")).hexdigest() + tools = create_parlant_tools( + self.features, + legacy_defaults=False, + additional_tools=self._custom_tools, + ) + guideline = await self._parlant_agent.create_guideline( + condition="Any incoming Band room message is being handled", + action=( + "Follow the Band platform instructions in this guideline. " + "Communicate through band_send_message with the intended " + "recipient in mentions; use Band tools for participants, " + "contacts, and memory when those tools are available. Treat the " + "initial @mention as the trigger target, not automatically as " + "the reply recipient." + ), + description=self._system_prompt, + matcher=p.MATCH_ALWAYS, + tools=tools, + metadata={ + "band_adapter_contract": True, + "prompt_hash": prompt_hash, + }, + ) + self._contract_guideline_id = str(guideline.id) + self._contract_guideline_installed = True + async def on_message( self, msg: PlatformMessage, @@ -155,8 +198,10 @@ async def on_message( logger.debug("Handling message %s in room %s", msg.id, room_id) if not self._app: - logger.error("Parlant Application not initialized") - return + error = "Parlant Application not initialized" + logger.error(error) + await self._report_error(tools, error) + raise RuntimeError(error) app = self._app sender_name = msg.sender_name or msg.sender_id or "User" @@ -167,11 +212,17 @@ async def on_message( except Exception as e: logger.error("Failed to get/create session for room %s: %s", room_id, e) await self._report_error(tools, f"Session initialization failed: {e}") - return + raise session_id_str = str(session_id) - # Set tools for this session (keyed by session_id for cross-task access) - set_session_tools(session_id_str, tools) + # Set tools for this session (keyed by session_id for cross-task access). + # Execution reporting is emitted in real time from the tool wrappers when + # enabled, so the wrappers need to know whether emit is on. + set_session_tools( + session_id_str, + tools, + emit_execution=Emit.EXECUTION in self.features.emit, + ) logger.info("Room %s: Set tools for session_id=%s", room_id, session_id_str) # On bootstrap, inject historical context @@ -276,10 +327,16 @@ async def _get_or_create_customer( if room_id in self._room_customers: return self._room_customers[room_id] - # Create customer via server + customer_id = f"band-{hashlib.sha256(room_id.encode('utf-8')).hexdigest()[:32]}" + + existing_customer = await self._server.find_customer(id=customer_id) + if existing_customer is not None: + self._room_customers[room_id] = existing_customer.id + return existing_customer.id + customer = await self._server.create_customer( name=customer_name, - id=f"band-{room_id[:8]}", + id=customer_id, ) self._room_customers[room_id] = customer.id @@ -384,32 +441,24 @@ async def _process_agent_response( tools: AgentToolsProtocol, sender_name: str, ) -> None: - """Wait for and process agent response events. - - Parlant may send multiple messages: - 1. A preamble message (tagged with __preamble__) - acknowledgment before tool execution - 2. Final message(s) after tool execution - - If the send_message tool was called during processing, we don't need to - forward Parlant's response (it would be a duplicate or empty). - """ + """Wait for and process agent response events.""" if not self._app: - logger.error("Room %s: No Parlant Application available", room_id) - return + raise RuntimeError(f"Room {room_id}: No Parlant Application available") app = self._app session_id_str = str(session_id) from parlant.core.async_utils import Timeout # type: ignore[missing-import] from parlant.core.sessions import EventKind, EventSource # type: ignore[missing-import] - current_offset = min_offset - max_iterations = 10 # Safety limit to prevent infinite loops - iteration = 0 + # Tool execution is reported in real time by the tool wrappers, so the + # response loop only waits for the agent's final message. + event_kinds = [EventKind.MESSAGE] + event_source = EventSource.AI_AGENT - while iteration < max_iterations: - iteration += 1 + current_offset = min_offset + max_iterations = 10 - # Wait for agent response + for iteration in range(1, max_iterations + 1): logger.info( "Room %s: Waiting for agent response (min_offset=%s, iteration=%s)...", room_id, @@ -421,66 +470,55 @@ async def _process_agent_response( has_update = await app.sessions.wait_for_more_events( # pyrefly: ignore[missing-attribute] session_id=session_id, min_offset=current_offset + 1, - kinds=[EventKind.MESSAGE], - source=EventSource.AI_AGENT, - timeout=Timeout(120), # Increased timeout for tool execution - ) - logger.info( - "Room %s: wait_for_more_events returned: %s", room_id, has_update + kinds=event_kinds, + source=event_source, + timeout=Timeout(120), ) except Exception as e: - logger.error( - "Room %s: Error waiting for update: %s", - room_id, - e, - exc_info=True, - ) - # Check if message was sent via tool before giving up if was_message_sent(session_id_str): logger.info( - "Room %s: Message was sent via tool, error is acceptable", - room_id, + "Room %s: Message was sent via tool before wait error", room_id ) - return + return + raise RuntimeError(f"Room {room_id}: Error waiting for response") from e if not has_update: - # Timeout - but check if message was already sent via tool if was_message_sent(session_id_str): - logger.info( - "Room %s: Timeout but message was sent via tool, OK", - room_id, - ) + logger.info("Room %s: Timeout after successful tool send", room_id) return - logger.warning("Room %s: Timeout waiting for agent response", room_id) - return + raise TimeoutError( + f"Room {room_id}: Timeout waiting for agent response" + ) - # Get new events try: events = await app.sessions.find_events( session_id=session_id, min_offset=current_offset + 1, - source=EventSource.AI_AGENT, - kinds=[EventKind.MESSAGE], - trace_id=None, # Required by Parlant SDK v3.x + source=event_source, + kinds=event_kinds, + trace_id=None, ) - logger.info("Room %s: Found %s agent events", room_id, len(events)) except Exception as e: - logger.error( - "Room %s: Error finding events: %s", - room_id, - e, - exc_info=True, - ) - return + if was_message_sent(session_id_str): + logger.info( + "Room %s: Message was sent via tool before event lookup error", + room_id, + ) + return + raise RuntimeError( + f"Room {room_id}: Error finding response events" + ) from e if not events: - logger.warning( - "Room %s: No events found despite update signal", room_id + if was_message_sent(session_id_str): + return + raise RuntimeError( + f"Room {room_id}: No response events found despite update signal" ) - return - # Process events and track if we got a non-preamble message - got_final_message = False + delivered = False + saw_non_preamble = False + saw_relevant_event = False for event in events: logger.debug( @@ -491,106 +529,82 @@ async def _process_agent_response( event.data, ) - # Update offset for next iteration if hasattr(event, "offset") and event.offset > current_offset: current_offset = event.offset if ( - event.kind == EventKind.MESSAGE - and event.source == EventSource.AI_AGENT + event.kind != EventKind.MESSAGE + or event.source != EventSource.AI_AGENT ): - data = event.data - message_content = "" - tags: list[str] = [] - - if isinstance(data, dict): - message_content = str(data.get("message", "")) - raw_tags = data.get("tags", []) - if isinstance(raw_tags, list): - tags = [str(tag) for tag in raw_tags] - elif isinstance(data, str): - message_content = data - - # Check if this is a preamble message - is_preamble = PARLANT_PREAMBLE_TAG in tags - - if is_preamble: - logger.info( - "Room %s: Skipping preamble message: %s...", - room_id, - message_content[:50], - ) - continue - - # Check if message was already sent via the send_message tool - # If so, don't send Parlant's response (would be duplicate/empty) - # Also don't mark as final - Parlant may still have more tool calls - if was_message_sent(session_id_str): - logger.info( - "Room %s: Message already sent via tool, skipping Parlant response: %s...", - room_id, - message_content[:50], - ) - continue - - # This is a final message (Parlant generated a response, not via tool) - got_final_message = True - - if message_content: - logger.info( - "Room %s: Sending agent response to platform: %s...", - room_id, - message_content[:100], - ) - try: - await tools.send_message( - message_content, mentions=[sender_name] - ) - logger.info("Room %s: Message sent successfully", room_id) - except Exception as e: - logger.error( - "Room %s: Error sending message: %s", - room_id, - e, - exc_info=True, - ) - else: - logger.warning( - "Room %s: Empty message content in event", - room_id, - ) - - # If we got a final (non-preamble) message, we're done - if got_final_message: - logger.info("Room %s: Got final message, processing complete", room_id) + continue + + saw_relevant_event = True + data = event.data + message_content = "" + tags: list[str] = [] + + if isinstance(data, dict): + message_content = str(data.get("message", "")) + raw_tags = data.get("tags", []) + if isinstance(raw_tags, list): + tags = [str(tag) for tag in raw_tags] + elif isinstance(data, str): + message_content = data + + if PARLANT_PREAMBLE_TAG in tags: + logger.info( + "Room %s: Skipping preamble message: %s...", + room_id, + message_content[:50], + ) + continue + + saw_non_preamble = True + + if was_message_sent(session_id_str): + logger.info( + "Room %s: Message already sent via tool, skipping Parlant response: %s...", + room_id, + message_content[:50], + ) + delivered = True + continue + + if not message_content: + raise RuntimeError( + f"Room {room_id}: Empty final message content from Parlant" + ) + + await tools.send_message(message_content, mentions=[sender_name]) + logger.info("Room %s: Message sent successfully", room_id) + delivered = True + + if delivered or was_message_sent(session_id_str): + logger.info("Room %s: Response delivery confirmed", room_id) return - # Check if message was sent via tool (tool execution may happen without final message) - if was_message_sent(session_id_str): + if saw_non_preamble: + raise RuntimeError( + f"Room {room_id}: Parlant produced a response but nothing was delivered" + ) + + if saw_relevant_event: logger.info( - "Room %s: Message sent via tool, no need to wait for final message", + "Room %s: Only got non-final events, continuing to wait for final message...", + room_id, + ) + else: + logger.info( + "Room %s: No relevant agent events found, continuing to wait...", room_id, ) - return - - # Otherwise, continue waiting for the final message after tool execution - logger.info( - "Room %s: Only got preamble, continuing to wait for final message...", - room_id, - ) - # Reached max iterations - check if message was sent if was_message_sent(session_id_str): - logger.info( - "Room %s: Max iterations but message was sent via tool, OK", - room_id, - ) - else: - logger.warning( - "Room %s: Reached max iterations (%s) waiting for response", - room_id, - max_iterations, - ) + logger.info("Room %s: Max iterations after successful tool send", room_id) + return + raise TimeoutError( + f"Room {room_id}: Reached max iterations ({max_iterations}) waiting for response" + ) async def on_cleanup(self, room_id: str) -> None: """Clean up session when agent leaves a room.""" diff --git a/src/band/integrations/parlant/tools.py b/src/band/integrations/parlant/tools.py index cd845c741..1b8403375 100644 --- a/src/band/integrations/parlant/tools.py +++ b/src/band/integrations/parlant/tools.py @@ -4,78 +4,120 @@ These tools are defined at server startup and use a session-keyed registry to access the current room's tools during execution. -This module provides the same tools as LangGraph/Claude adapters: -- send_message: Send messages to the chat room -- send_event: Send events (thought, error, task) -- add_participant: Add agents/users to the room -- remove_participant: Remove participants -- lookup_peers: Find available agents -- get_participants: List current participants -- create_chatroom: Create new rooms -- list_contacts: List agent's contacts -- add_contact: Send a contact request -- remove_contact: Remove an existing contact -- list_contact_requests: List received and sent requests -- respond_contact_request: Approve, reject, or cancel requests - NOTE: We intentionally do NOT use `from __future__ import annotations` here because Parlant's @p.tool decorator checks annotation types at runtime. """ +from dataclasses import dataclass +import inspect import json import logging +from types import UnionType +from typing import Any, Optional, Union, get_args, get_origin +import uuid import warnings -from typing import Any, Optional +from pydantic_core import PydanticUndefined + +from band.core.exceptions import BandToolError +from band.core.protocols import AgentToolsProtocol +from band.core.tool_filter import filter_tool_schemas from band.core.types import AdapterFeatures, Capability +from band.runtime.custom_tools import ( + CustomToolDef, + execute_custom_tool, + get_custom_tool_name, +) +from band.runtime.tools import ( + CONTACT_TOOL_NAMES, + MEMORY_TOOL_NAMES, + ToolDefinition, + get_tool_description, + iter_tool_definitions, +) logger = logging.getLogger(__name__) -# Session-keyed registry to hold tools for each session -# This approach works across async contexts (unlike ContextVar) -_session_tools: dict[str, Any] = {} -# Track whether send_message was called for each session -# This helps the adapter know if it needs to forward Parlant's response -_session_message_sent: dict[str, bool] = {} +@dataclass +class _SessionContext: + tools: AgentToolsProtocol + message_sent: bool = False + emit_execution: bool = False + + +# Session-keyed registry to hold tools and delivery state. +# This approach works across async contexts (unlike ContextVar). +_session_contexts: dict[str, _SessionContext] = {} +# Platform tools already create user-visible Band effects directly, so they are +# not re-reported as execution events (that would double-count the send). +_SILENT_REPORTING_TOOLS = frozenset({"band_send_message", "band_send_event"}) -def set_session_tools(session_id: str, tools: Optional[Any]) -> None: - """Set the tools for a specific Parlant session.""" +_ERROR_PREFIXES = ( + "Error", + "Invalid arguments", + "Unknown tool", +) + + +def set_session_tools( + session_id: str, + tools: Optional[AgentToolsProtocol], + *, + emit_execution: bool = False, +) -> None: + """Set the tools for a specific Parlant session. + + Args: + session_id: Parlant session ID the tools belong to. + tools: Room AgentTools, or ``None`` to clear the session. + emit_execution: When true, generated tool wrappers report each call as + ``tool_call``/``tool_result`` Band events in real time, interleaved + with the actual side effects. + """ if tools is None: - _session_tools.pop(session_id, None) - _session_message_sent.pop(session_id, None) + _session_contexts.pop(session_id, None) else: - _session_tools[session_id] = tools - _session_message_sent[session_id] = False + _session_contexts[session_id] = _SessionContext( + tools=tools, + emit_execution=emit_execution, + ) logger.debug("Set tools for session %s: %s", session_id, tools is not None) -def get_session_tools(session_id: str) -> Optional[Any]: - """Get the tools for a specific Parlant session.""" - tools = _session_tools.get(session_id) +def _get_session_context(session_id: str) -> Optional[_SessionContext]: + context = _session_contexts.get(session_id) logger.debug( - "Get tools for session_id=%s: found=%s, available_sessions=%s", + "Get context for session_id=%s: found=%s, available_sessions=%s", session_id, - tools is not None, - list(_session_tools.keys()), + context is not None, + list(_session_contexts.keys()), ) - return tools + return context + + +def get_session_tools(session_id: str) -> Optional[AgentToolsProtocol]: + """Get the tools for a specific Parlant session.""" + context = _get_session_context(session_id) + return context.tools if context else None def mark_message_sent(session_id: str) -> None: """Mark that a message was sent via the send_message tool for this session.""" - _session_message_sent[session_id] = True + if context := _get_session_context(session_id): + context.message_sent = True logger.debug("Marked message sent for session %s", session_id) def was_message_sent(session_id: str) -> bool: - """Check if a message was sent via the send_message tool for this session.""" - return _session_message_sent.get(session_id, False) + """Check if a message was sent via the send_message tool.""" + context = _get_session_context(session_id) + return bool(context and context.message_sent) -# Keep old API for backwards compatibility (deprecated) -def set_current_tools(tools: Optional[Any]) -> None: +# Keep old API for backwards compatibility (deprecated). +def set_current_tools(tools: Optional[AgentToolsProtocol]) -> None: """Deprecated: Use set_session_tools instead.""" warnings.warn( "set_current_tools is deprecated, use set_session_tools instead", @@ -84,627 +126,399 @@ def set_current_tools(tools: Optional[Any]) -> None: ) -def get_current_tools() -> Optional[Any]: +def get_current_tools() -> Optional[AgentToolsProtocol]: """Deprecated: Use get_session_tools instead.""" warnings.warn( "get_current_tools is deprecated, use get_session_tools instead", DeprecationWarning, stacklevel=2, ) - return None # Always returns None, tools now accessed via session_id + return None -def create_parlant_tools(features: AdapterFeatures | None = None) -> list[Any]: - """Create Parlant tool definitions that wrap Band tools. +def _tool_name(entry: Any) -> str: + return str(entry.tool.name) - These tools use context variables to access the current room's - AgentToolsProtocol during execution. - Args: - features: Optional adapter features. When CONTACTS capability is absent, - contact-management tools are excluded from the returned list. +def _tool_category(entry: Any) -> str | None: + name = _tool_name(entry) + if name in MEMORY_TOOL_NAMES: + return "memory" + if name in CONTACT_TOOL_NAMES: + return "contacts" + return "chat" - Returns: - List of Parlant ToolEntry objects - """ - try: - import parlant.sdk as p # type: ignore[missing-import] - from parlant.core.tools import ToolContext, ToolResult # type: ignore[missing-import] - except ImportError: - logger.warning("Parlant SDK not installed, skipping tool creation") - return [] - @p.tool - async def band_send_message( - context: ToolContext, - content: str, - mentions: str, - ) -> ToolResult: - """ - Send a message to the chat room. - - Use this to respond to users or other agents. Messages require @mentions - to reach users. You MUST use this tool to communicate. - - Args: - context: Parlant tool context (automatically provided) - content: The message content to send - mentions: Comma-separated list of participant handles to @mention (e.g., "@alice, @bob/agent") - - Returns: - Confirmation of message sent or error - """ - logger.info( - "[Parlant Tool] send_message called: session=%s, content=%s..., mentions=%s", - context.session_id, - content[:50], - mentions, - ) - tools = get_session_tools(context.session_id) - if not tools: - logger.error( - "[Parlant Tool] send_message: No tools available for session %s", - context.session_id, - ) - return ToolResult(data="Error: No tools available in current context") +def _is_error_result(result: Any) -> bool: + return isinstance(result, str) and result.startswith(_ERROR_PREFIXES) + + +def _dump_data(result: Any) -> Any: + """Normalize Fern/Pydantic models for JSON-friendly tool output.""" + if hasattr(result, "model_dump"): + return result.model_dump() + if isinstance(result, list): + return [_dump_data(item) for item in result] + if isinstance(result, dict): + return {key: _dump_data(value) for key, value in result.items()} + return result + + +def _tool_result_data(result: Any) -> str: + if isinstance(result, str): + return result + return json.dumps(_dump_data(result), default=str) + + +def _is_union(annotation: Any) -> bool: + return get_origin(annotation) in (Union, UnionType) + +def _is_optional(annotation: Any) -> bool: + return _is_union(annotation) and type(None) in get_args(annotation) + + +def _optional_inner(annotation: Any) -> Any: + return next(arg for arg in get_args(annotation) if arg is not type(None)) + + +def _is_literal(annotation: Any) -> bool: + return str(get_origin(annotation)) == "typing.Literal" + + +def _is_dict_annotation(annotation: Any) -> bool: + origin = get_origin(annotation) + if origin is dict: + return True + if _is_optional(annotation): + return _is_dict_annotation(_optional_inner(annotation)) + return False + + +def _parlant_supported_annotation(annotation: Any) -> Any: + """Map Pydantic field annotations to Parlant-supported tool annotations.""" + if _is_optional(annotation): + inner = _parlant_supported_annotation(_optional_inner(annotation)) + return inner | None + if _is_literal(annotation) or annotation is Any: + return str + if _is_dict_annotation(annotation): + return str + return annotation + + +def _coerce_parlant_arguments( + tool_name: str, + input_model: type[Any], + arguments: dict[str, Any], +) -> dict[str, Any] | str: + """Convert Parlant-friendly values back to canonical tool arguments.""" + coerced = dict(arguments) + for name, field in input_model.model_fields.items(): + if name not in coerced or coerced[name] in (None, ""): + continue + if not _is_dict_annotation(field.annotation): + continue + if not isinstance(coerced[name], str): + continue try: - # Parse mentions from comma-separated string - mention_list = [m.strip() for m in mentions.split(",") if m.strip()] - if not mention_list: - logger.warning("[Parlant Tool] send_message: No mentions provided") - return ToolResult(data="Error: At least one mention is required") - - logger.info("[Parlant Tool] Sending message to: %s", mention_list) - await tools.send_message(content, mention_list) - # Mark that we sent a message via the tool (so adapter doesn't duplicate) - mark_message_sent(context.session_id) - logger.info("[Parlant Tool] Message sent successfully via tool") - return ToolResult(data=f"Message sent to {', '.join(mention_list)}") - except Exception as e: - logger.error("[Parlant Tool] Error sending message: %s", e, exc_info=True) - return ToolResult(data=f"Error sending message: {e}") - - @p.tool - async def band_send_event( - context: ToolContext, - content: str, - message_type: str, - ) -> ToolResult: - """ - Send an event to the chat room. No mentions required. - - Use this to share your reasoning or report status. - - Args: - context: Parlant tool context (automatically provided) - content: Human-readable event content - message_type: Type of event - 'thought' (share reasoning), 'error' (report problem), or 'task' (report progress) - - Returns: - Confirmation of event sent or error - """ - logger.info( - "[Parlant Tool] send_event called: session=%s, type=%s", - context.session_id, - message_type, - ) - tools = get_session_tools(context.session_id) - if not tools: - logger.error( - "[Parlant Tool] send_event: No tools available for session %s", - context.session_id, + coerced[name] = json.loads(coerced[name]) + except json.JSONDecodeError as error: + return ( + f"Invalid arguments for {tool_name}: {name} must be valid JSON: {error}" ) - return ToolResult(data="Error: No tools available in current context") + return coerced + + +def _signature_for_input_model( + input_model: type[Any], + tool_context_type: type[Any], + tool_result_type: type[Any], +) -> inspect.Signature: + parameters = [ + inspect.Parameter( + "context", + inspect.Parameter.POSITIONAL_OR_KEYWORD, + annotation=tool_context_type, + ) + ] - if message_type not in ("thought", "error", "task"): - return ToolResult( - data=f"Error: Invalid message_type '{message_type}'. Use 'thought', 'error', or 'task'" + for name, field in input_model.model_fields.items(): + default = inspect.Parameter.empty + if not field.is_required(): + default = None if field.default is PydanticUndefined else field.default + + parameters.append( + inspect.Parameter( + name, + inspect.Parameter.POSITIONAL_OR_KEYWORD, + default=default, + annotation=_parlant_supported_annotation(field.annotation), ) + ) - try: - await tools.send_event(content, message_type, None) - logger.info("[Parlant Tool] Event (%s) sent successfully", message_type) - return ToolResult(data=f"Event ({message_type}) sent successfully") - except Exception as e: - logger.error("[Parlant Tool] Error sending event: %s", e, exc_info=True) - return ToolResult(data=f"Error sending event: {e}") - - @p.tool - async def band_add_participant( - context: ToolContext, - identifier: str, - ) -> ToolResult: - """ - Invite an agent or user to join this chat room. - - Args: - context: Parlant tool context (automatically provided) - identifier: REQUIRED - Handle, name, or ID of the agent to add. Prefer the exact ID returned by lookup_peers; handles are mainly for mentions. Use lookup_peers to find available agents. - - Returns: - Success message or error description - - Example calls: - add_participant(identifier="pirate-captain") - add_participant(identifier="Research Agent") - """ - logger.info( - "[Parlant Tool] add_participant called: session=%s, identifier=%s", - context.session_id, - identifier, + return inspect.Signature( + parameters=parameters, + return_annotation=tool_result_type, + ) + + +def _new_tool_call_id() -> str: + return f"parlant-{uuid.uuid4().hex[:12]}" + + +async def _report_tool_call( + session_context: _SessionContext, + tool_name: str, + arguments: Any, + tool_call_id: str, +) -> None: + """Emit a Band ``tool_call`` event in real time, before the tool runs.""" + if not session_context.emit_execution or tool_name in _SILENT_REPORTING_TOOLS: + return + try: + await session_context.tools.send_event( + content=json.dumps( + { + "name": tool_name, + "args": _dump_data(arguments), + "tool_call_id": tool_call_id, + }, + default=str, + ), + message_type="tool_call", ) - tools = get_session_tools(context.session_id) - if not tools: - logger.error( - "[Parlant Tool] add_participant: No tools available for session %s", - context.session_id, - ) - return ToolResult(data="Error: No tools available in current context") + except Exception as error: + logger.warning("Failed to report tool_call for %s: %s", tool_name, error) + + +async def _report_tool_result( + session_context: _SessionContext, + tool_name: str, + output: Any, + tool_call_id: str, +) -> None: + """Emit a Band ``tool_result`` event in real time, after the tool runs.""" + if not session_context.emit_execution or tool_name in _SILENT_REPORTING_TOOLS: + return + try: + await session_context.tools.send_event( + content=json.dumps( + { + "name": tool_name, + "output": _dump_data(output), + "tool_call_id": tool_call_id, + }, + default=str, + ), + message_type="tool_result", + ) + except Exception as error: + logger.warning("Failed to report tool_result for %s: %s", tool_name, error) + + +def _create_builtin_parlant_tool_entry( + definition: ToolDefinition, + p: Any, + tool_context_type: type[Any], + tool_result_type: type[Any], +) -> Any: + signature = _signature_for_input_model( + definition.input_model, + tool_context_type, + tool_result_type, + ) + + async def wrapper(context: Any, *args: Any, **kwargs: Any) -> Any: + bound = signature.bind(context, *args, **kwargs) + bound.apply_defaults() + arguments = { + key: value for key, value in bound.arguments.items() if key != "context" + } + session_id = str(context.session_id) + session_context = _get_session_context(session_id) + + if not session_context: + return tool_result_type(data="Error: No tools available in current context") + + coerced = _coerce_parlant_arguments( + definition.name, + definition.input_model, + arguments, + ) + if isinstance(coerced, str): + return tool_result_type(data=coerced) + + tool_call_id = _new_tool_call_id() + await _report_tool_call(session_context, definition.name, coerced, tool_call_id) try: - result = await tools.add_participant(identifier, "member") - status = result.get("status", "added") - if status == "already_in_room": - logger.info("[Parlant Tool] '%s' is already in the room", identifier) - return ToolResult( - data=f"'{identifier}' is already in the room - no action needed" - ) - logger.info( - "[Parlant Tool] Successfully added '%s' to the room", identifier + result = await session_context.tools.execute_tool_call( + definition.name, + coerced, ) - return ToolResult(data=f"Successfully added '{identifier}' to the room") - except Exception as e: + except BandToolError as error: logger.error( - "[Parlant Tool] Error adding participant '%s': %s", - identifier, - e, + "[Parlant Tool] Band tool error in %s: %s", + definition.name, + error, exc_info=True, ) - return ToolResult(data=f"Error adding participant '{identifier}': {e}") - - @p.tool - async def band_remove_participant( - context: ToolContext, - identifier: str, - ) -> ToolResult: - """ - Remove a participant from this chat room. - - Args: - context: Parlant tool context (automatically provided) - identifier: REQUIRED - Handle, name, or ID of the participant to remove. - - Returns: - Success message or error description - - Example calls: - remove_participant(identifier="pirate-captain") - remove_participant(identifier="Research Agent") - """ - logger.info( - "[Parlant Tool] remove_participant called: session=%s, identifier=%s", - context.session_id, - identifier, - ) - tools = get_session_tools(context.session_id) - if not tools: - logger.error( - "[Parlant Tool] remove_participant: No tools available for session %s", - context.session_id, - ) - return ToolResult(data="Error: No tools available in current context") - - try: - await tools.remove_participant(identifier) - logger.info( - "[Parlant Tool] Successfully removed '%s' from the room", identifier + output = f"Error executing {definition.name}: {error}" + await _report_tool_result( + session_context, definition.name, output, tool_call_id ) - return ToolResult(data=f"Successfully removed '{identifier}' from the room") - except Exception as e: + return tool_result_type(data=output) + except Exception as error: logger.error( - "[Parlant Tool] Error removing participant '%s': %s", - identifier, - e, + "[Parlant Tool] Unexpected error in %s: %s", + definition.name, + error, exc_info=True, ) - return ToolResult(data=f"Error removing participant '{identifier}': {e}") - - @p.tool - async def band_lookup_peers( - context: ToolContext, - ) -> ToolResult: - """ - List available peers (agents and users) that can be added to this room. - - Automatically excludes peers already in the room. Use this to find - specialized agents when you cannot answer a question directly. - - Args: - context: Parlant tool context (automatically provided) - - Returns: - List of available agents with their names and descriptions - """ - logger.info( - "[Parlant Tool] lookup_peers called: session=%s", context.session_id - ) - tools = get_session_tools(context.session_id) - if not tools: - logger.error( - "[Parlant Tool] lookup_peers: No tools available for session %s", - context.session_id, + output = f"Error executing {definition.name}: {error}" + await _report_tool_result( + session_context, definition.name, output, tool_call_id ) - return ToolResult(data="Error: No tools available in current context") + return tool_result_type(data=output) - try: - # Use defaults - pagination rarely needed for agent lookups - result = await tools.lookup_peers(page=1, page_size=50) - logger.info("[Parlant Tool] lookup_peers result: %s", result) - # Normalize Fern model -> dict for uniform handling - data = result.model_dump() if hasattr(result, "model_dump") else result - peers = data.get("data") or data.get("peers") or [] - metadata = data.get("metadata") or {} - if not peers: - return ToolResult(data="No available agents found") - - page_num = metadata.get("page", 1) - total_pages = metadata.get("total_pages", 1) - lines = [f"Available agents (page {page_num} of {total_pages}):"] - for peer in peers: - name = peer.get("name", "Unknown") - desc = peer.get("description") or "No description" - peer_type = peer.get("type", "Agent") - lines.append(f"- {name} ({peer_type}): {desc}") - return ToolResult(data="\n".join(lines)) - except Exception as e: - logger.error("[Parlant Tool] Error looking up peers: %s", e, exc_info=True) - return ToolResult(data=f"Error looking up peers: {e}") - - @p.tool - async def band_get_participants( - context: ToolContext, - ) -> ToolResult: - """ - Get the list of all participants currently in the chat room. - - Args: - context: Parlant tool context (automatically provided) - - Returns: - List of current participants with their names and types - """ - logger.info( - "[Parlant Tool] get_participants called: session=%s", context.session_id - ) - tools = get_session_tools(context.session_id) - if not tools: - logger.error( - "[Parlant Tool] get_participants: No tools available for session %s", - context.session_id, - ) - return ToolResult(data="Error: No tools available in current context") + if definition.name == "band_send_message" and not _is_error_result(result): + session_context.message_sent = True - try: - result = await tools.get_participants() - logger.info("[Parlant Tool] get_participants result: %s", result) - # Normalize Fern models -> dicts for uniform handling - if isinstance(result, list): - items = [ - p.model_dump() if hasattr(p, "model_dump") else p for p in result - ] - if not items: - return ToolResult(data="No participants in the room") - lines = ["Current participants:"] - for participant in items: - name = participant.get("name", "Unknown") - p_type = participant.get("type", "Unknown") - lines.append(f"- {name} ({p_type})") - return ToolResult(data="\n".join(lines)) - return ToolResult(data=str(result)) - except Exception as e: - logger.error( - "[Parlant Tool] Error getting participants: %s", e, exc_info=True - ) - return ToolResult(data=f"Error getting participants: {e}") - - @p.tool - async def band_create_chatroom( - context: ToolContext, - task_id: str = "", - ) -> ToolResult: - """ - Create a new chat room for a specific task or conversation. - - Args: - context: Parlant tool context (automatically provided) - task_id: Optional task ID to associate with the room - - Returns: - The ID of the newly created room - """ - logger.info( - "[Parlant Tool] create_chatroom called: session=%s, task_id=%s", - context.session_id, - task_id, + await _report_tool_result( + session_context, definition.name, result, tool_call_id ) - tools = get_session_tools(context.session_id) - if not tools: - logger.error( - "[Parlant Tool] create_chatroom: No tools available for session %s", - context.session_id, - ) - return ToolResult(data="Error: No tools available in current context") + output = _tool_result_data(result) + return tool_result_type(data=output) + + wrapper.__name__ = definition.name + wrapper.__qualname__ = definition.name + wrapper.__doc__ = get_tool_description(definition.name) + wrapper.__signature__ = signature # type: ignore[attr-defined] + + return p.tool(name=definition.name)(wrapper) + + +def _create_custom_parlant_tool_entry( + custom_tool: CustomToolDef, + p: Any, + tool_context_type: type[Any], + tool_result_type: type[Any], +) -> Any: + input_model, _handler = custom_tool + tool_name = get_custom_tool_name(input_model) + signature = _signature_for_input_model( + input_model, + tool_context_type, + tool_result_type, + ) - try: - result = await tools.create_chatroom(task_id if task_id else None) - logger.info("[Parlant Tool] Created chatroom: %s", result) - return ToolResult(data=f"Created new chat room: {result}") - except Exception as e: - logger.error("[Parlant Tool] Error creating chatroom: %s", e, exc_info=True) - return ToolResult(data=f"Error creating chatroom: {e}") - - include_contacts = features is None or Capability.CONTACTS in features.capabilities - - @p.tool - async def band_list_contacts( - context: ToolContext, - page: int = 1, - page_size: int = 50, - ) -> ToolResult: - """ - List agent's contacts with pagination. - - Args: - context: Parlant tool context (automatically provided) - page: Page number (default 1) - page_size: Items per page (default 50, max 100) - - Returns: - JSON with contacts list and pagination metadata - """ - logger.info( - "[Parlant Tool] list_contacts called: session=%s, page=%s", - context.session_id, - page, - ) - tools = get_session_tools(context.session_id) - if not tools: - logger.error( - "[Parlant Tool] list_contacts: No tools available for session %s", - context.session_id, - ) - return ToolResult(data="Error: No tools available in current context") + async def wrapper(context: Any, *args: Any, **kwargs: Any) -> Any: + bound = signature.bind(context, *args, **kwargs) + bound.apply_defaults() + arguments = { + key: value for key, value in bound.arguments.items() if key != "context" + } + coerced = _coerce_parlant_arguments(tool_name, input_model, arguments) + if isinstance(coerced, str): + return tool_result_type(data=coerced) + + session_context = _get_session_context(str(context.session_id)) + tool_call_id = _new_tool_call_id() + if session_context: + await _report_tool_call(session_context, tool_name, coerced, tool_call_id) try: - result = await tools.list_contacts(page, page_size) - # Fern model: serialize via model_dump if available, fallback to str - data = result.model_dump() if hasattr(result, "model_dump") else result - return ToolResult(data=json.dumps(data, default=str)) - except Exception as e: - logger.error("[Parlant Tool] Error listing contacts: %s", e, exc_info=True) - return ToolResult(data=f"Error listing contacts: {e}") - - @p.tool - async def band_add_contact( - context: ToolContext, - handle: str, - message: str = "", - ) -> ToolResult: - """ - Send a contact request to add someone as a contact. - - Returns 'pending' when request is created, 'approved' when inverse - request existed and was auto-accepted. - - Args: - context: Parlant tool context (automatically provided) - handle: Handle of user/agent to add (e.g., '@john' or '@john/agent-name') - message: Optional message with the request - - Returns: - Status of the contact request - """ - logger.info( - "[Parlant Tool] add_contact called: session=%s, handle=%s", - context.session_id, - handle, - ) - tools = get_session_tools(context.session_id) - if not tools: + result = await execute_custom_tool(custom_tool, coerced) + except ValueError as error: + output = str(error) + if session_context: + await _report_tool_result( + session_context, tool_name, output, tool_call_id + ) + return tool_result_type(data=output) + except Exception as error: logger.error( - "[Parlant Tool] add_contact: No tools available for session %s", - context.session_id, + "[Parlant Tool] Unexpected error in custom tool %s: %s", + tool_name, + error, + exc_info=True, ) - return ToolResult(data="Error: No tools available in current context") + output = f"Error executing {tool_name}: {error}" + if session_context: + await _report_tool_result( + session_context, tool_name, output, tool_call_id + ) + return tool_result_type(data=output) - try: - result = await tools.add_contact(handle, message if message else None) - data = result.model_dump() if hasattr(result, "model_dump") else result - status = ( - data.get("status", "pending") if isinstance(data, dict) else "pending" - ) - return ToolResult(data=f"Contact request to {handle}: {status}") - except Exception as e: - logger.error("[Parlant Tool] Error adding contact: %s", e, exc_info=True) - return ToolResult(data=f"Error adding contact: {e}") - - @p.tool - async def band_remove_contact( - context: ToolContext, - handle: str = "", - contact_id: str = "", - ) -> ToolResult: - """ - Remove an existing contact by handle or contact ID. - - Provide either handle or contact_id (at least one required). - - Args: - context: Parlant tool context (automatically provided) - handle: Contact's handle (e.g., '@john') - contact_id: Or contact record ID (UUID) - - Returns: - Confirmation of contact removal - """ - logger.info( - "[Parlant Tool] remove_contact called: session=%s, handle=%s, contact_id=%s", - context.session_id, - handle, - contact_id, - ) - tools = get_session_tools(context.session_id) - if not tools: - logger.error( - "[Parlant Tool] remove_contact: No tools available for session %s", - context.session_id, - ) - return ToolResult(data="Error: No tools available in current context") + if session_context: + await _report_tool_result(session_context, tool_name, result, tool_call_id) + output = _tool_result_data(result) + return tool_result_type(data=output) - h = handle if handle else None - cid = contact_id if contact_id else None - if not h and not cid: - return ToolResult( - data="Error: Either handle or contact_id must be provided" - ) + wrapper.__name__ = tool_name + wrapper.__qualname__ = tool_name + wrapper.__doc__ = input_model.__doc__ or f"Execute {tool_name}" + wrapper.__signature__ = signature # type: ignore[attr-defined] - try: - await tools.remove_contact(h, cid) - identifier = handle or contact_id - return ToolResult(data=f"Contact '{identifier}' removed successfully") - except Exception as e: - logger.error("[Parlant Tool] Error removing contact: %s", e, exc_info=True) - return ToolResult(data=f"Error removing contact: {e}") - - @p.tool - async def band_list_contact_requests( - context: ToolContext, - page: int = 1, - page_size: int = 50, - sent_status: str = "pending", - ) -> ToolResult: - """ - List both received and sent contact requests. - - Received requests are always filtered to pending status. - Sent requests can be filtered by status. - - Args: - context: Parlant tool context (automatically provided) - page: Page number (default 1) - page_size: Items per page per direction (default 50, max 100) - sent_status: Filter sent requests by status: 'pending', 'approved', 'rejected', 'cancelled', or 'all' - - Returns: - JSON with received and sent request lists and metadata - """ - logger.info( - "[Parlant Tool] list_contact_requests called: session=%s, sent_status=%s", - context.session_id, - sent_status, - ) - tools = get_session_tools(context.session_id) - if not tools: - logger.error( - "[Parlant Tool] list_contact_requests: No tools available for session %s", - context.session_id, - ) - return ToolResult(data="Error: No tools available in current context") + return p.tool(name=tool_name)(wrapper) - try: - result = await tools.list_contact_requests(page, page_size, sent_status) - # Fern model: serialize via model_dump if available, fallback to str - data = result.model_dump() if hasattr(result, "model_dump") else result - return ToolResult(data=json.dumps(data, default=str)) - except Exception as e: - logger.error( - "[Parlant Tool] Error listing contact requests: %s", e, exc_info=True - ) - return ToolResult(data=f"Error listing contact requests: {e}") - - @p.tool - async def band_respond_contact_request( - context: ToolContext, - action: str, - handle: str = "", - request_id: str = "", - ) -> ToolResult: - """ - Respond to a contact request. - - Actions: - - 'approve'/'reject': For requests you RECEIVED (handle = requester's handle) - - 'cancel': For requests you SENT (handle = recipient's handle) - - Provide either handle or request_id (at least one required). - - Args: - context: Parlant tool context (automatically provided) - action: Action to take - 'approve', 'reject', or 'cancel' - handle: Other party's handle - request_id: Or request ID (UUID) - - Returns: - Status of the response action - """ - logger.info( - "[Parlant Tool] respond_contact_request called: session=%s, action=%s", - context.session_id, - action, - ) - tools = get_session_tools(context.session_id) - if not tools: - logger.error( - "[Parlant Tool] respond_contact_request: No tools available for session %s", - context.session_id, - ) - return ToolResult(data="Error: No tools available in current context") - h = handle if handle else None - rid = request_id if request_id else None - if not h and not rid: - return ToolResult( - data="Error: Either handle or request_id must be provided" - ) +def create_parlant_tools( + features: AdapterFeatures | None = None, + *, + legacy_defaults: bool | None = None, + additional_tools: list[CustomToolDef] | None = None, +) -> list[Any]: + """Create Parlant tool definitions that wrap canonical Band tools. - if action not in ("approve", "reject", "cancel"): - return ToolResult( - data=f"Error: Invalid action '{action}'. Use 'approve', 'reject', or 'cancel'" - ) + Args: + features: Optional adapter features. Explicit features control contact + and memory capability exposure. + legacy_defaults: When true, preserve the historical direct-call default + of exposing contact tools even when no explicit feature selection was + provided. Defaults to true only when ``features`` is ``None``. + Adapter code passes this based on whether the caller supplied + ``features=``. + additional_tools: CustomToolDef tuples to expose as native Parlant tools. - try: - result = await tools.respond_contact_request(action, h, rid) - data = result.model_dump() if hasattr(result, "model_dump") else result - status = data.get("status", action) if isinstance(data, dict) else action - return ToolResult(data=f"Contact request {action}d: {status}") - except Exception as e: - logger.error( - "[Parlant Tool] Error responding to contact request: %s", - e, - exc_info=True, - ) - return ToolResult(data=f"Error responding to contact request: {e}") - - tools = [ - band_send_message, - band_send_event, - band_add_participant, - band_remove_participant, - band_lookup_peers, - band_get_participants, - band_create_chatroom, - ] + Returns: + List of Parlant ToolEntry objects. + """ + try: + import parlant.sdk as p # type: ignore[missing-import] + from parlant.core.tools import ToolContext, ToolResult # type: ignore[missing-import] + except ImportError: + logger.warning("Parlant SDK not installed, skipping tool creation") + return [] - if include_contacts: - tools.extend( - [ - band_list_contacts, - band_add_contact, - band_remove_contact, - band_list_contact_requests, - band_respond_contact_request, - ] + feature_config = features or AdapterFeatures() + use_legacy_defaults = ( + features is None if legacy_defaults is None else legacy_defaults + ) + include_contacts = ( + True + if use_legacy_defaults + else Capability.CONTACTS in feature_config.capabilities + ) + include_memory = Capability.MEMORY in feature_config.capabilities + + entries = [ + _create_builtin_parlant_tool_entry(definition, p, ToolContext, ToolResult) + for definition in iter_tool_definitions( + surface="agent", + include_memory=include_memory, + include_contacts=include_contacts, ) + ] + entries.extend( + _create_custom_parlant_tool_entry(tool, p, ToolContext, ToolResult) + for tool in additional_tools or [] + ) - return tools + return filter_tool_schemas( + entries, + feature_config, + get_name=_tool_name, + get_category=_tool_category, + ) diff --git a/tests/adapters/test_parlant_adapter.py b/tests/adapters/test_parlant_adapter.py index 0fa4cfd53..37a63e1bd 100644 --- a/tests/adapters/test_parlant_adapter.py +++ b/tests/adapters/test_parlant_adapter.py @@ -12,9 +12,36 @@ import sys import pytest +from pydantic import BaseModel + +try: + import parlant.sdk # type: ignore[missing-import] # noqa: F401 + + _HAS_PARLANT = True +except ImportError: + _HAS_PARLANT = False from band.adapters.parlant import ParlantAdapter -from band.core.types import PlatformMessage +from band.core.types import AdapterFeatures, Capability, Emit, PlatformMessage + +# Parlant lives in the isolated `dev-parlant` dependency fork (it conflicts with +# crewai). The plain `test` CI job installs `--extra dev` without parlant and +# runs the whole suite, so these parlant-specific tests must skip cleanly there; +# they run for real in the `test-parlant` job. +pytestmark = pytest.mark.skipif( + not _HAS_PARLANT, + reason="parlant not installed (uv sync --extra dev-parlant)", +) + + +class CalculatorInput(BaseModel): + """Calculate a value.""" + + value: int + + +def calculate(args: CalculatorInput) -> str: + return str(args.value + 1) @pytest.fixture @@ -64,7 +91,8 @@ def mock_parlant_server(): # Container returns Application server.container = {MagicMock: mock_app} - # Mock create_customer + # Mock customer lookup/creation + server.find_customer = AsyncMock(return_value=None) server.create_customer = AsyncMock(return_value=MagicMock(id="customer-123")) return server @@ -76,6 +104,7 @@ def mock_parlant_agent(): agent = MagicMock() agent.id = "parlant-agent-123" agent.name = "TestBot" + agent.create_guideline = AsyncMock(return_value=MagicMock(id="guideline-123")) return agent @@ -106,6 +135,20 @@ def test_internal_state_initialized(self, mock_parlant_server, mock_parlant_agen assert adapter._room_customers == {} assert adapter._system_prompt == "" + def test_stores_additional_tools_for_contract_guideline( + self, mock_parlant_server, mock_parlant_agent + ): + """Parlant should expose CustomToolDef tools through its guideline tools.""" + custom_tool = (CalculatorInput, calculate) + + adapter = ParlantAdapter( + server=mock_parlant_server, + parlant_agent=mock_parlant_agent, + additional_tools=[custom_tool], + ) + + assert adapter._custom_tools == [custom_tool] + class TestOnStarted: """Tests for on_started() method.""" @@ -171,6 +214,117 @@ async def test_uses_custom_system_prompt_if_provided( assert adapter._system_prompt == "You are a custom assistant." + @pytest.mark.asyncio + async def test_installs_rendered_prompt_as_parlant_guideline( + self, mock_parlant_server, mock_parlant_agent, mock_application_class + ): + """Should install Band platform instructions into Parlant, not just store them.""" + adapter = ParlantAdapter( + server=mock_parlant_server, + parlant_agent=mock_parlant_agent, + custom_section="Always include the token BANANA.", + features=AdapterFeatures(capabilities={Capability.CONTACTS}), + ) + + mock_app = MagicMock() + mock_module = MagicMock() + mock_module.Application = mock_application_class + mock_parlant_server.container = {mock_application_class: mock_app} + + with patch.dict( + sys.modules, + {"parlant.core.application": mock_module}, + ): + await adapter.on_started( + agent_name="BandBot", agent_description="A Band test agent" + ) + + mock_parlant_agent.create_guideline.assert_awaited_once() + kwargs = mock_parlant_agent.create_guideline.await_args.kwargs + assert "BANANA" in kwargs["description"] + assert "BandBot" in kwargs["description"] + assert kwargs["metadata"]["band_adapter_contract"] is True + assert kwargs["matcher"] is not None + assert any(t.tool.name == "band_send_message" for t in kwargs["tools"]) + assert any(t.tool.name == "band_list_contacts" for t in kwargs["tools"]) + assert adapter._contract_guideline_installed is True + assert adapter._contract_guideline_id == "guideline-123" + + @pytest.mark.asyncio + async def test_contract_guideline_includes_additional_tools( + self, mock_parlant_server, mock_parlant_agent, mock_application_class + ): + """CustomToolDef tools should be exposed through Parlant's tool surface.""" + adapter = ParlantAdapter( + server=mock_parlant_server, + parlant_agent=mock_parlant_agent, + additional_tools=[(CalculatorInput, calculate)], + ) + + mock_app = MagicMock() + mock_module = MagicMock() + mock_module.Application = mock_application_class + mock_parlant_server.container = {mock_application_class: mock_app} + + with patch.dict( + sys.modules, + {"parlant.core.application": mock_module}, + ): + await adapter.on_started("BandBot", "A Band test agent") + + tools = mock_parlant_agent.create_guideline.await_args.kwargs["tools"] + calculator = next(t for t in tools if t.tool.name == "calculator") + assert list(calculator.tool.parameters) == ["value"] + + @pytest.mark.asyncio + async def test_contract_guideline_requires_explicit_contacts_capability( + self, mock_parlant_server, mock_parlant_agent, mock_application_class + ): + """Omitted features should not expose contact-management tools.""" + adapter = ParlantAdapter( + server=mock_parlant_server, + parlant_agent=mock_parlant_agent, + ) + + mock_app = MagicMock() + mock_module = MagicMock() + mock_module.Application = mock_application_class + mock_parlant_server.container = {mock_application_class: mock_app} + + with patch.dict( + sys.modules, + {"parlant.core.application": mock_module}, + ): + await adapter.on_started("BandBot", "A Band test agent") + + tools = mock_parlant_agent.create_guideline.await_args.kwargs["tools"] + assert not any(t.tool.name == "band_list_contacts" for t in tools) + + @pytest.mark.asyncio + async def test_contract_guideline_respects_explicit_empty_features( + self, mock_parlant_server, mock_parlant_agent, mock_application_class + ): + """Explicit empty features should not expose contact tools.""" + adapter = ParlantAdapter( + server=mock_parlant_server, + parlant_agent=mock_parlant_agent, + features=AdapterFeatures(), + ) + + mock_app = MagicMock() + mock_module = MagicMock() + mock_module.Application = mock_application_class + mock_parlant_server.container = {mock_application_class: mock_app} + + with patch.dict( + sys.modules, + {"parlant.core.application": mock_module}, + ): + await adapter.on_started("BandBot", "A Band test agent") + + tools = mock_parlant_agent.create_guideline.await_args.kwargs["tools"] + assert not any(t.tool.name == "band_list_contacts" for t in tools) + @pytest.mark.asyncio async def test_gets_application_from_container( self, mock_parlant_server, mock_parlant_agent, mock_application_class @@ -218,8 +372,17 @@ def initialized_adapter(self, mock_parlant_server, mock_parlant_agent): mock_app.sessions.create_customer_message = AsyncMock( return_value=MagicMock(offset=1) ) - mock_app.sessions.wait_for_update = AsyncMock(return_value=True) - mock_app.sessions.find_events = AsyncMock(return_value=[]) + mock_app.sessions.wait_for_more_events = AsyncMock(return_value=True) + mock_app.sessions.find_events = AsyncMock( + return_value=[ + MagicMock( + offset=2, + kind="message", + source="ai_agent", + data={"message": "Hello from Parlant"}, + ) + ] + ) adapter._app = mock_app return adapter @@ -237,7 +400,8 @@ async def test_creates_session_for_room( Moderation=MagicMock(NONE="none") ), "parlant.core.sessions": MagicMock( - EventSource=MagicMock(CUSTOMER="customer") + EventSource=MagicMock(CUSTOMER="customer", AI_AGENT="ai_agent"), + EventKind=MagicMock(MESSAGE="message"), ), "parlant.core.async_utils": MagicMock(Timeout=lambda x: x), }, @@ -256,6 +420,33 @@ async def test_creates_session_for_room( assert "room-123" in initialized_adapter._room_sessions mock_parlant_server.create_customer.assert_called_once() + @pytest.mark.asyncio + async def test_customer_ids_do_not_truncate_room_id_prefixes( + self, initialized_adapter, mock_parlant_server + ): + """Rooms with the same first eight characters should not share a customer id.""" + await initialized_adapter._get_or_create_customer("abcdefgh-room-one", "Alice") + await initialized_adapter._get_or_create_customer("abcdefgh-room-two", "Bob") + + first_call, second_call = mock_parlant_server.create_customer.await_args_list + assert first_call.kwargs["id"] != second_call.kwargs["id"] + + @pytest.mark.asyncio + async def test_get_or_create_customer_reuses_existing_parlant_customer( + self, initialized_adapter, mock_parlant_server + ): + """Adapter restarts against the same Parlant server should be idempotent.""" + mock_parlant_server.find_customer.return_value = MagicMock( + id="existing-customer" + ) + + customer_id = await initialized_adapter._get_or_create_customer( + "room-123", "Alice" + ) + + assert customer_id == "existing-customer" + mock_parlant_server.create_customer.assert_not_called() + @pytest.mark.asyncio async def test_sends_customer_message_to_parlant( self, initialized_adapter, sample_message, mock_tools @@ -293,6 +484,12 @@ async def test_sends_customer_message_to_parlant( # Verify message was sent to Parlant initialized_adapter._app.sessions.create_customer_message.assert_called_once() + wait_kwargs = ( + initialized_adapter._app.sessions.wait_for_more_events.await_args.kwargs + ) + find_kwargs = initialized_adapter._app.sessions.find_events.await_args.kwargs + assert wait_kwargs["source"] == "ai_agent" + assert find_kwargs["source"] == "ai_agent" @pytest.mark.asyncio async def test_sets_session_tools_for_tool_execution( @@ -328,11 +525,158 @@ async def test_sets_session_tools_for_tool_execution( # Verify tools were set with session_id and then cleared assert mock_set_tools.call_count == 2 - # First call sets the tools with session_id - mock_set_tools.assert_any_call("session-123", mock_tools) + # First call sets the tools with session_id + emit flag (off by default) + mock_set_tools.assert_any_call( + "session-123", mock_tools, emit_execution=False + ) # Second call clears the tools mock_set_tools.assert_any_call("session-123", None) + @pytest.mark.asyncio + async def test_response_loop_only_waits_for_agent_message( + self, mock_parlant_server, mock_parlant_agent, sample_message, mock_tools + ): + """Execution reporting moved into the tool wrappers, so the response loop + only waits for the agent's final MESSAGE — it no longer polls TOOL events + or widens the source filter, which is what produced misordered/duplicate + tool events.""" + adapter = ParlantAdapter( + server=mock_parlant_server, + parlant_agent=mock_parlant_agent, + features=AdapterFeatures(emit={Emit.EXECUTION}), + ) + adapter.agent_name = "TestBot" + adapter.agent_description = "A test bot" + adapter._system_prompt = "Test prompt" + + mock_app = MagicMock() + mock_app.sessions = AsyncMock() + mock_app.sessions.create = AsyncMock(return_value=MagicMock(id="session-123")) + mock_app.sessions.create_customer_message = AsyncMock( + return_value=MagicMock(offset=1) + ) + mock_app.sessions.wait_for_more_events = AsyncMock(return_value=True) + mock_app.sessions.find_events = AsyncMock( + return_value=[ + MagicMock( + id="evt-message", + offset=3, + kind="message", + source="ai_agent", + data={"message": "Done"}, + ), + ] + ) + adapter._app = mock_app + + mock_moderation = MagicMock() + mock_moderation.NONE = "none" + mock_event_kind = MagicMock(MESSAGE="message", TOOL="tool") + mock_event_source = MagicMock( + CUSTOMER="customer", + AI_AGENT="ai_agent", + SYSTEM="system", + ) + + with patch.dict( + sys.modules, + { + "parlant.core.app_modules.sessions": MagicMock( + Moderation=mock_moderation + ), + "parlant.core.sessions": MagicMock( + EventSource=mock_event_source, + EventKind=mock_event_kind, + ), + "parlant.core.async_utils": MagicMock(Timeout=lambda x: x), + }, + ): + await adapter.on_message( + msg=sample_message, + tools=mock_tools, + history=[], + participants_msg=None, + contacts_msg=None, + is_session_bootstrap=True, + room_id="room-123", + ) + + wait_kwargs = mock_app.sessions.wait_for_more_events.await_args.kwargs + find_kwargs = mock_app.sessions.find_events.await_args.kwargs + # Only the final agent message is awaited; TOOL polling is gone. + assert wait_kwargs["kinds"] == ["message"] + assert find_kwargs["kinds"] == ["message"] + assert wait_kwargs["source"] == "ai_agent" + assert find_kwargs["source"] == "ai_agent" + # The adapter no longer re-reports tool events from the session log. + mock_tools.send_event.assert_not_called() + mock_tools.send_message.assert_awaited_once_with("Done", mentions=["Alice"]) + + @pytest.mark.asyncio + async def test_emit_flag_passed_to_session_tools( + self, mock_parlant_server, mock_parlant_agent, sample_message, mock_tools + ): + """The adapter must tell the wrappers whether execution emit is enabled.""" + adapter = ParlantAdapter( + server=mock_parlant_server, + parlant_agent=mock_parlant_agent, + features=AdapterFeatures(emit={Emit.EXECUTION}), + ) + adapter.agent_name = "TestBot" + adapter.agent_description = "A test bot" + adapter._system_prompt = "Test prompt" + + mock_app = MagicMock() + mock_app.sessions = AsyncMock() + mock_app.sessions.create = AsyncMock(return_value=MagicMock(id="session-123")) + mock_app.sessions.create_customer_message = AsyncMock( + return_value=MagicMock(offset=1) + ) + mock_app.sessions.wait_for_more_events = AsyncMock(return_value=True) + mock_app.sessions.find_events = AsyncMock( + return_value=[ + MagicMock( + id="evt-message", + offset=3, + kind="message", + source="ai_agent", + data={"message": "Done"}, + ), + ] + ) + adapter._app = mock_app + + mock_moderation = MagicMock() + mock_moderation.NONE = "none" + mock_event_kind = MagicMock(MESSAGE="message", TOOL="tool") + mock_event_source = MagicMock(CUSTOMER="customer", AI_AGENT="ai_agent") + + with patch.dict( + sys.modules, + { + "parlant.core.app_modules.sessions": MagicMock( + Moderation=mock_moderation + ), + "parlant.core.sessions": MagicMock( + EventSource=mock_event_source, + EventKind=mock_event_kind, + ), + "parlant.core.async_utils": MagicMock(Timeout=lambda x: x), + }, + ): + with patch("band.adapters.parlant.set_session_tools") as mock_set_tools: + await adapter.on_message( + msg=sample_message, + tools=mock_tools, + history=[], + participants_msg=None, + contacts_msg=None, + is_session_bootstrap=True, + room_id="room-123", + ) + + mock_set_tools.assert_any_call("session-123", mock_tools, emit_execution=True) + @pytest.mark.asyncio async def test_reuses_existing_session( self, initialized_adapter, sample_message, mock_tools, mock_parlant_server @@ -595,16 +939,16 @@ async def test_handles_uninitialized_app( ) # Don't set _app - # Should return early without error - await adapter.on_message( - msg=sample_message, - tools=mock_tools, - history=[], - participants_msg=None, - contacts_msg=None, - is_session_bootstrap=True, - room_id="room-123", - ) + with pytest.raises(RuntimeError, match="Parlant Application not initialized"): + await adapter.on_message( + msg=sample_message, + tools=mock_tools, + history=[], + participants_msg=None, + contacts_msg=None, + is_session_bootstrap=True, + room_id="room-123", + ) - # No calls should be made mock_tools.send_message.assert_not_called() + mock_tools.send_event.assert_called_once() diff --git a/tests/e2e/adapters/test_parlant.py b/tests/e2e/adapters/test_parlant.py index 7d6a80543..385088c18 100644 --- a/tests/e2e/adapters/test_parlant.py +++ b/tests/e2e/adapters/test_parlant.py @@ -13,18 +13,26 @@ from __future__ import annotations -from collections.abc import AsyncGenerator +from collections.abc import AsyncGenerator, Callable +import asyncio +import contextlib +import json +import os +import uuid import pytest -from band_rest import AsyncRestClient +from pydantic import BaseModel +from band_rest import AsyncRestClient, ChatRoomRequest +from band_rest.types import ParticipantRequest from band.agent import Agent +from band.core.types import AdapterFeatures, Emit from tests.e2e.conftest import E2ESettings, requires_e2e from tests.e2e.helpers import ( TrackingWebSocketClient, run_smoke_test, - run_tool_execution_test, + send_trigger_message, ) try: @@ -34,9 +42,131 @@ except ImportError: HAS_PARLANT = False + +class NativeEchoInput(BaseModel): + """Echo back the validation code provided by the user.""" + + code: str + + +def native_echo_handler(args: NativeEchoInput) -> dict: + """Custom additional_tool used to exercise the wrapper's real-time emission.""" + return {"echo": f"verified-{args.code}"} + + requires_parlant = pytest.mark.skipif(not HAS_PARLANT, reason="parlant not installed") +def _message_value(payload, key: str): + if isinstance(payload, dict): + return payload.get(key) + return getattr(payload, key, None) + + +def _is_agent_text_message(payload, agent_id: str, expected_content: str) -> bool: + return ( + _message_value(payload, "message_type") == "text" + and _message_value(payload, "sender_type") == "Agent" + and _message_value(payload, "sender_id") == agent_id + and expected_content in str(_message_value(payload, "content") or "") + ) + + +def _message_timestamp_key(payload) -> str: + timestamp = _message_value(payload, "inserted_at") or _message_value( + payload, "created_at" + ) + if timestamp is None: + raise AssertionError("Chat message is missing inserted_at/created_at timestamp") + return str(timestamp) + + +async def _wait_for_chat_messages( + client: AsyncRestClient, + chat_id: str, + predicate: Callable[[list], bool], + timeout: float, +): + """Poll the durable human-visible room history until the expected messages exist.""" + deadline = asyncio.get_running_loop().time() + timeout + last_messages = [] + while asyncio.get_running_loop().time() < deadline: + response = await client.human_api_messages.list_my_chat_messages( + chat_id, + page_size=50, + ) + last_messages = list(response.data or []) + if predicate(last_messages): + return last_messages + await asyncio.sleep(0.5) + + summary = [ + { + "type": _message_value(msg, "message_type"), + "sender": _message_value(msg, "sender_name"), + "content": str(_message_value(msg, "content") or "")[:160], + } + for msg in last_messages[:12] + ] + raise TimeoutError(f"Timed out waiting for expected Parlant messages: {summary}") + + +def _configure_parlant_agent_credentials() -> None: + """Use the local Parlant agent config when a sourced env exposes a user key.""" + current_key = os.getenv("BAND_API_KEY", "") + if current_key.startswith(("thnv_a", "band_a")) and os.getenv("TEST_AGENT_ID"): + return + + if current_key.startswith(("thnv_u", "band_u")) and not os.getenv( + "BAND_API_KEY_USER" + ): + os.environ["BAND_API_KEY_USER"] = current_key + + try: + from band.config import load_agent_config + + agent_id, api_key = load_agent_config("tom_agent") + except (FileNotFoundError, ValueError): + return + + os.environ["BAND_API_KEY"] = api_key + os.environ["TEST_AGENT_ID"] = agent_id + os.environ.setdefault("BAND_AGENT_ID", agent_id) + + +_configure_parlant_agent_credentials() + + +@pytest.fixture +async def e2e_parlant_room( + e2e_session_client: AsyncRestClient, + e2e_created_room_ids: list[str], +) -> tuple[str, str, str]: + """Create a fresh Band room for each Parlant E2E test. + + Parlant E2E starts a new in-process Parlant server per test. Reusing a + persistent Band room would hydrate stale prompts and responses into that new + Parlant session, making LLM behavior depend on previous runs. + """ + peers_response = await e2e_session_client.agent_api_peers.list_agent_peers() + user_peer = next((p for p in peers_response.data if p.type == "User"), None) + if user_peer is None: + pytest.skip("No User peer available for Parlant E2E tests") + + response = await e2e_session_client.agent_api_chats.create_agent_chat( + chat=ChatRoomRequest() + ) + if response.data is None: + pytest.fail("create_agent_chat returned no data") + room_id = response.data.id + await e2e_session_client.agent_api_participants.add_agent_chat_participant( + room_id, + participant=ParticipantRequest(participant_id=user_peer.id, role="member"), + ) + e2e_created_room_ids.append(room_id) + return room_id, user_peer.id, user_peer.name + + @pytest.mark.asyncio @requires_e2e @requires_parlant @@ -58,16 +188,53 @@ async def running_parlant_agent( """ from band.adapters.parlant import ParlantAdapter - async with p.Server() as server: + if not os.getenv("OPENAI_API_KEY"): + pytest.skip("OPENAI_API_KEY is required for Parlant E2E tests") + + server = p.Server( + host="127.0.0.1", + port=0, + tool_service_port=0, + nlp_service=p.NLPServices.openai, + ) + await server.__aenter__() + try: parlant_agent = await server.create_agent( name="E2E Test Agent", - description="A test agent for E2E validation. Keep responses short.", + description=( + "A test agent for E2E validation. Keep responses short. " + "Incoming messages start with a mention to you; treat that as " + "the trigger target, not the reply recipient. Reply to the " + "user who sent the message." + ), + ) + # Steering guidelines only. Band/built-in tools and the custom + # `nativeecho` tool are registered by the adapter's contract + # guideline via ``create_parlant_tools(additional_tools=...)``. + await parlant_agent.create_guideline( + condition="User asks you to reply with a specific word or phrase", + action=( + "Call band_send_message with the requested word or phrase " + "as content, and set mentions to the user's name or handle. " + "Do not address or mention yourself." + ), + ) + await parlant_agent.create_guideline( + condition="User asks for echo validation with a code", + action=( + "Call the nativeecho tool with the exact validation code as " + "the `code` argument. Then call band_send_message with " + "content containing the returned echo code, and mention the " + "user, not yourself." + ), ) adapter = ParlantAdapter( server=server, parlant_agent=parlant_agent, custom_section="Keep responses short and concise.", + features=AdapterFeatures(emit={Emit.EXECUTION}), + additional_tools=[(NativeEchoInput, native_echo_handler)], ) agent = Agent.create( @@ -80,8 +247,10 @@ async def running_parlant_agent( async with agent: yield agent + finally: + with contextlib.suppress(TimeoutError): + await asyncio.wait_for(server.__aexit__(None, None, None), timeout=30) - @pytest.mark.flaky(reruns=2) async def test_smoke_responds_to_message( self, e2e_config: E2ESettings, @@ -105,7 +274,6 @@ async def test_smoke_responds_to_message( adapter_name="parlant", ) - @pytest.mark.flaky(reruns=2) async def test_tool_execution_send_message( self, e2e_config: E2ESettings, @@ -116,15 +284,108 @@ async def test_tool_execution_send_message( api_client: AsyncRestClient, ): """Verify the agent uses band_send_message tool to respond.""" + chat_id, _user_id, user_name = e2e_parlant_room + agent_id, agent_name = e2e_agent_info + token = f"PINEAPPLE-{uuid.uuid4().hex[:8]}" + await send_trigger_message( + api_client, + chat_id, + f"Reply to {user_name} with the exact phrase {token}. Do not reply to {agent_name}.", + agent_name, + agent_id, + ) + received = await _wait_for_chat_messages( + api_client, + chat_id, + lambda messages: any( + _is_agent_text_message(msg, agent_id, token) for msg in messages + ), + e2e_config.e2e_timeout, + ) + + assert any(_is_agent_text_message(msg, agent_id, token) for msg in received) + + async def test_execution_emit_reports_additional_custom_tool( + self, + e2e_config: E2ESettings, + e2e_parlant_room: tuple[str, str, str], + e2e_agent_info: tuple[str, str], + ws_client: TrackingWebSocketClient, + running_parlant_agent: Agent, + api_client: AsyncRestClient, + ): + """Emit.EXECUTION reports adapter-registered tools (built-ins + additional_tools) + in real time, with tool_call ordered before tool_result and before the reply. + + Raw Parlant tools attached directly to a guideline outside additional_tools + are intentionally not reported by the wrapper-based path. + """ chat_id, _user_id, _user_name = e2e_parlant_room agent_id, agent_name = e2e_agent_info + code = f"NATIVE-{uuid.uuid4().hex[:8]}" + expected_echo = f"verified-{code}" - await run_tool_execution_test( - ws_client, + def has_expected_messages(messages) -> bool: + has_tool_call = any( + _message_value(msg, "message_type") == "tool_call" + and "nativeecho" in str(_message_value(msg, "content") or "") + and code in str(_message_value(msg, "content") or "") + for msg in messages + ) + has_tool_result = any( + _message_value(msg, "message_type") == "tool_result" + and "nativeecho" in str(_message_value(msg, "content") or "") + and code in str(_message_value(msg, "content") or "") + for msg in messages + ) + has_text_reply = any( + _is_agent_text_message(msg, agent_id, expected_echo) for msg in messages + ) + return has_tool_call and has_tool_result and has_text_reply + + await send_trigger_message( api_client, chat_id, + f"Echo validation: call the nativeecho tool with code {code}, then reply to the user with the returned echo value.", agent_name, agent_id, - timeout=e2e_config.e2e_timeout, - adapter_name="parlant", ) + received = await _wait_for_chat_messages( + api_client, + chat_id, + has_expected_messages, + e2e_config.e2e_timeout, + ) + + tool_call = next( + msg + for msg in received + if _message_value(msg, "message_type") == "tool_call" + and "nativeecho" in str(_message_value(msg, "content") or "") + and code in str(_message_value(msg, "content") or "") + ) + tool_result = next( + msg + for msg in received + if _message_value(msg, "message_type") == "tool_result" + and "nativeecho" in str(_message_value(msg, "content") or "") + and code in str(_message_value(msg, "content") or "") + ) + text_reply = next( + msg + for msg in received + if _is_agent_text_message(msg, agent_id, expected_echo) + ) + ordered_messages = sorted(received, key=_message_timestamp_key) + assert ordered_messages.index(tool_call) < ordered_messages.index(tool_result) + assert ordered_messages.index(tool_result) < ordered_messages.index(text_reply) + + call_payload = json.loads(_message_value(tool_call, "content")) + result_payload = json.loads(_message_value(tool_result, "content")) + assert call_payload["name"] == "nativeecho" + assert call_payload["args"]["code"] == code + assert result_payload["name"] == "nativeecho" + assert result_payload["output"]["echo"] == expected_echo + # tool_call and tool_result are correlated by a stable id emitted by the + # wrapper around a single execute_custom_tool invocation. + assert call_payload["tool_call_id"] == result_payload["tool_call_id"] diff --git a/tests/framework_configs/adapters.py b/tests/framework_configs/adapters.py index 61574447e..775868f76 100644 --- a/tests/framework_configs/adapters.py +++ b/tests/framework_configs/adapters.py @@ -12,7 +12,7 @@ import threading from dataclasses import dataclass, field from typing import Any, Callable -from unittest.mock import MagicMock +from unittest.mock import AsyncMock, MagicMock from tests.framework_configs._sentinel import MISSING, STRICT_CI, _MissingSentinel from band.adapters.claude_sdk import _CLAUDE_SDK_AVAILABLE as _HAS_CLAUDE_SDK @@ -242,6 +242,9 @@ def _parlant_factory(**kw: Any) -> Any: mock_agent = MagicMock() mock_agent.id = "parlant-agent-123" mock_agent.name = "TestBot" + mock_agent.create_guideline = AsyncMock( + return_value=MagicMock(id="guideline-123") + ) kw["parlant_agent"] = mock_agent return ParlantAdapter(**kw) @@ -573,7 +576,8 @@ def _build_parlant_config() -> AdapterConfig: "system_prompt": "Custom system prompt", "custom_section": "Be helpful.", }, - has_custom_tools_attr=False, + has_custom_tools_attr=True, + custom_tools_attr="_custom_tools", # on_started does a runtime `from parlant.core.application import Application` # which fails when parlant SDK is not installed (conflict group with crewai). skip_on_started_conformance=not _parlant_available, diff --git a/tests/framework_conformance/test_tool_name_drift.py b/tests/framework_conformance/test_tool_name_drift.py index 7cd829f50..2dc3c1f5a 100644 --- a/tests/framework_conformance/test_tool_name_drift.py +++ b/tests/framework_conformance/test_tool_name_drift.py @@ -35,6 +35,13 @@ if _HAS_CLAUDE_SDK: from band.integrations.claude_sdk.tools import build_band_sdk_tools +try: + import parlant.sdk # noqa: F401 + + _HAS_PARLANT = True +except ImportError: + _HAS_PARLANT = False + _SRC_ROOT = Path(__file__).resolve().parents[2] / "src" / "band" if not _SRC_ROOT.is_dir(): raise FileNotFoundError(f"Source root not found: {_SRC_ROOT}") @@ -90,7 +97,7 @@ def test_derives_memory_tools_from_central_registry(self): reason="claude-agent-sdk not installed (pip install band-sdk[claude_sdk])", ) def test_shared_builder_covers_all_tools(self): - """Every Band tool should be buildable for the Claude SDK adapter.""" + """Every Band platform tool should be buildable for the Claude SDK adapter.""" sdk_tools = build_band_sdk_tools( tool_definitions=iter_tool_definitions(include_memory=True), get_tools=lambda _room_id: None, @@ -109,7 +116,7 @@ class TestClaudeSDKIntegrationToolDrift: _FILE = _SRC_ROOT / "integrations" / "claude_sdk" / "tools.py" def test_derives_tool_list_from_central_registry(self): - """Verify BAND_CHAT_TOOLS is derived from CHAT_TOOL_NAMES.""" + """Verify chat tools are derived from CHAT_TOOL_NAMES.""" source = self._FILE.read_text() assert "CHAT_TOOL_NAMES" in source, ( "Claude SDK integration tools should import CHAT_TOOL_NAMES from " @@ -224,18 +231,44 @@ def test_supports_memory_tools_toggle(self): class TestParlantToolDrift: - """Parlant integration (integrations/parlant/tools.py) — chat tools only.""" + """Parlant integration (integrations/parlant/tools.py) — chat tools only. + + The Parlant integration derives its tool list dynamically from the central + registry via ``iter_tool_definitions`` instead of declaring one hand-written + ``@p.tool`` function per tool, so individual tool names no longer appear as + string literals in the source. We verify the dynamic wiring statically and, + when Parlant is installed, that the built tool set actually covers every + chat tool. + """ _FILE = _SRC_ROOT / "integrations" / "parlant" / "tools.py" - def test_all_chat_tools_registered(self): - """Every chat tool has a Parlant tool function.""" + def test_derives_tools_from_central_registry(self): + """Verify tools are built from iter_tool_definitions, not hardcoded.""" source = self._FILE.read_text() - found = _extract_tool_names(source) + assert "iter_tool_definitions" in source, ( + "Parlant integration should derive its tool list from " + "iter_tool_definitions() in band.runtime.tools instead of " + "hardcoding one tool function per tool name." + ) + + @pytest.mark.skipif( + not _HAS_PARLANT, + reason="parlant not installed (uv sync --extra dev-parlant)", + ) + def test_all_chat_tools_registered(self): + """Every chat tool is built by create_parlant_tools().""" + from band.integrations.parlant.tools import ( + _tool_name, + create_parlant_tools, + ) + + entries = create_parlant_tools() + found = {_tool_name(entry) for entry in entries} missing = CHAT_TOOL_NAMES - found assert not missing, ( f"Parlant integration is missing tool functions for: {sorted(missing)}. " - f"Add tool implementations in create_parlant_tools()." + f"create_parlant_tools() must build every chat tool from the registry." ) diff --git a/tests/integrations/parlant/test_tools.py b/tests/integrations/parlant/test_tools.py index 7a9dc84e4..cc0c6a8d4 100644 --- a/tests/integrations/parlant/test_tools.py +++ b/tests/integrations/parlant/test_tools.py @@ -1,18 +1,47 @@ """Tests for Parlant tools module.""" +import json +from types import SimpleNamespace from unittest.mock import AsyncMock, MagicMock import pytest +from pydantic import BaseModel +from band.core.types import AdapterFeatures, Capability from band.integrations.parlant.tools import ( - _session_message_sent, - _session_tools, + _session_contexts, create_parlant_tools, get_session_tools, mark_message_sent, set_session_tools, was_message_sent, ) +from band.runtime.tools import iter_tool_definitions + +try: + import parlant.sdk # type: ignore[missing-import] # noqa: F401 + + _HAS_PARLANT = True +except ImportError: + _HAS_PARLANT = False + +# create_parlant_tools() returns real wrappers only when parlant is importable. +# Parlant lives in the isolated `dev-parlant` fork, so skip cleanly in the plain +# `test` job and run for real in `test-parlant`. +pytestmark = pytest.mark.skipif( + not _HAS_PARLANT, + reason="parlant not installed (uv sync --extra dev-parlant)", +) + + +class CalculatorInput(BaseModel): + """Add one to the provided value.""" + + value: int + + +def calculate(args: CalculatorInput) -> str: + return str(args.value + 1) class TestSessionToolsRegistry: @@ -20,8 +49,7 @@ class TestSessionToolsRegistry: def setup_method(self): """Clear registry before each test.""" - _session_tools.clear() - _session_message_sent.clear() + _session_contexts.clear() def test_set_session_tools_stores_tools(self): """Should store tools for a session.""" @@ -29,8 +57,8 @@ def test_set_session_tools_stores_tools(self): set_session_tools("session-123", mock_tools) - assert "session-123" in _session_tools - assert _session_tools["session-123"] is mock_tools + assert "session-123" in _session_contexts + assert _session_contexts["session-123"].tools is mock_tools def test_set_session_tools_initializes_message_sent_flag(self): """Should initialize message_sent flag to False.""" @@ -38,23 +66,22 @@ def test_set_session_tools_initializes_message_sent_flag(self): set_session_tools("session-123", mock_tools) - assert _session_message_sent["session-123"] is False + assert _session_contexts["session-123"].message_sent is False def test_set_session_tools_clears_on_none(self): """Should clear tools when setting None.""" mock_tools = MagicMock() set_session_tools("session-123", mock_tools) - assert "session-123" in _session_tools + assert "session-123" in _session_contexts set_session_tools("session-123", None) - assert "session-123" not in _session_tools - assert "session-123" not in _session_message_sent + assert "session-123" not in _session_contexts def test_get_session_tools_returns_stored_tools(self): """Should return stored tools for session.""" mock_tools = MagicMock() - _session_tools["session-123"] = mock_tools + set_session_tools("session-123", mock_tools) result = get_session_tools("session-123") @@ -72,28 +99,28 @@ class TestMessageSentFlag: def setup_method(self): """Clear registry before each test.""" - _session_tools.clear() - _session_message_sent.clear() + _session_contexts.clear() def test_mark_message_sent_sets_flag(self): """Should set message_sent flag to True.""" - _session_message_sent["session-123"] = False + set_session_tools("session-123", MagicMock()) mark_message_sent("session-123") - assert _session_message_sent["session-123"] is True + assert _session_contexts["session-123"].message_sent is True def test_was_message_sent_returns_true_when_sent(self): - """Should return True when message was sent.""" - _session_message_sent["session-123"] = True + """Should return True when sent.""" + set_session_tools("session-123", MagicMock()) + mark_message_sent("session-123") result = was_message_sent("session-123") assert result is True def test_was_message_sent_returns_false_when_not_sent(self): - """Should return False when message was not sent.""" - _session_message_sent["session-123"] = False + """Should return False when not sent.""" + set_session_tools("session-123", MagicMock()) result = was_message_sent("session-123") @@ -124,7 +151,7 @@ def test_get_current_tools_emits_deprecation_warning(self): get_current_tools() def test_get_current_tools_returns_none(self): - """Should return None (tools now accessed via session_id).""" + """Should return None because tools are session-keyed now.""" from band.integrations.parlant.tools import get_current_tools with pytest.warns(DeprecationWarning): @@ -136,393 +163,442 @@ def test_get_current_tools_returns_none(self): class TestCreateParlantTools: """Tests for create_parlant_tools() function.""" - def test_returns_list_of_tools(self): - """Should return list of tool entries when Parlant is installed.""" + def test_returns_tool_entries_from_canonical_registry(self): + """Generated Parlant tools should come from runtime ToolDefinition entries.""" tools = create_parlant_tools() - assert isinstance(tools, list) - # Non-empty; specific tool names are verified in the next test. - # Avoid hardcoded counts so adding/removing tools doesn't silently - # break this assertion — the next test validates the exact contract. - assert len(tools) > 0 - - def test_returns_expected_tool_names(self): - """Should return tools with expected names.""" - tools = create_parlant_tools() - - # Tools are ToolEntry objects with a .tool attribute containing the Tool - tool_names = [t.tool.name for t in tools] - assert "band_send_message" in tool_names - assert "band_send_event" in tool_names - assert "band_add_participant" in tool_names - assert "band_remove_participant" in tool_names - assert "band_lookup_peers" in tool_names - assert "band_get_participants" in tool_names - assert "band_create_chatroom" in tool_names - assert "band_list_contacts" in tool_names - assert "band_add_contact" in tool_names - assert "band_remove_contact" in tool_names - assert "band_list_contact_requests" in tool_names - assert "band_respond_contact_request" in tool_names - - def test_tools_have_descriptions(self): - """Should have descriptions for all tools.""" + tool_names = [entry.tool.name for entry in tools] + expected_names = [ + definition.name + for definition in iter_tool_definitions( + surface="agent", + include_memory=False, + include_contacts=True, + ) + ] + assert tool_names == expected_names + + def test_generated_tools_have_canonical_descriptions(self): + """Every generated Parlant tool should expose the canonical description.""" tools = create_parlant_tools() for entry in tools: assert entry.tool.description, f"Tool {entry.tool.name} has no description" - def test_send_message_tool_has_required_parameters(self): - """send_message should have content and mentions parameters.""" - tools = create_parlant_tools() - - send_message_entry = next( - t for t in tools if t.tool.name == "band_send_message" + def test_generated_parameters_match_tool_models(self): + """Parlant signatures should not drift from canonical Pydantic tool models.""" + entries = {entry.tool.name: entry for entry in create_parlant_tools()} + + for definition in iter_tool_definitions( + surface="agent", + include_memory=False, + include_contacts=True, + ): + entry = entries[definition.name] + expected_params = list(definition.input_model.model_fields) + expected_required = [ + name + for name, field in definition.input_model.model_fields.items() + if field.is_required() + ] + + assert list(entry.tool.parameters) == expected_params + assert entry.tool.required == expected_required + + def test_send_message_mentions_parameter_is_array(self): + """The wrapper should expose mentions as the canonical list field.""" + entry = next( + entry + for entry in create_parlant_tools() + if entry.tool.name == "band_send_message" ) - # Parameters is a dict with param names as keys - param_names = list(send_message_entry.tool.parameters.keys()) - assert "content" in param_names - assert "mentions" in param_names + descriptor, _options = entry.tool.parameters["mentions"] + assert descriptor["type"] == "array" + assert descriptor["item_type"] == "string" - def test_send_event_tool_has_message_type_parameter(self): - """send_event should have message_type parameter.""" - tools = create_parlant_tools() + def test_excludes_contact_tools_without_capability(self): + """Explicit empty capabilities should exclude contact and memory tools.""" + tools = create_parlant_tools(features=AdapterFeatures()) + tool_names = [entry.tool.name for entry in tools] - send_event_entry = next(t for t in tools if t.tool.name == "band_send_event") - param_names = list(send_event_entry.tool.parameters.keys()) + assert "band_send_message" in tool_names + assert "band_list_contacts" not in tool_names + assert "band_store_memory" not in tool_names - assert "content" in param_names - assert "message_type" in param_names + def test_includes_contact_tools_with_capability(self): + """Contact tools are exposed when CONTACTS capability is enabled.""" + tools = create_parlant_tools( + features=AdapterFeatures(capabilities=frozenset({Capability.CONTACTS})) + ) + tool_names = [entry.tool.name for entry in tools] - def test_add_participant_tool_has_identifier_parameter(self): - """add_participant should have identifier parameter.""" - tools = create_parlant_tools() + assert "band_list_contacts" in tool_names + assert "band_add_contact" in tool_names + assert "band_store_memory" not in tool_names - add_participant_entry = next( - t for t in tools if t.tool.name == "band_add_participant" + def test_includes_memory_tools_with_capability(self): + """Memory tools are exposed when MEMORY capability is enabled.""" + tools = create_parlant_tools( + features=AdapterFeatures(capabilities=frozenset({Capability.MEMORY})) ) - param_names = list(add_participant_entry.tool.parameters.keys()) + tool_names = [entry.tool.name for entry in tools] - assert "identifier" in param_names + assert "band_store_memory" in tool_names + assert "band_get_memory" in tool_names + assert "band_list_contacts" not in tool_names - def test_lookup_peers_has_no_parameters(self): - """lookup_peers should have no user-facing parameters (pagination is hardcoded).""" + def test_includes_contact_tools_when_no_features(self): + """Legacy default direct calls should still expose contact tools.""" tools = create_parlant_tools() + tool_names = [entry.tool.name for entry in tools] - lookup_peers_entry = next( - t for t in tools if t.tool.name == "band_lookup_peers" + assert "band_list_contacts" in tool_names + assert "band_store_memory" not in tool_names + + def test_include_tools_filters_by_parlant_tool_name(self): + """include_tools should narrow generated Parlant ToolEntry objects.""" + tools = create_parlant_tools( + features=AdapterFeatures(include_tools=frozenset({"band_send_event"})) ) - param_names = list(lookup_peers_entry.tool.parameters.keys()) - # Pagination was intentionally removed to simplify the API - # The function uses hardcoded defaults (page=1, page_size=50) - assert param_names == [] + assert [entry.tool.name for entry in tools] == ["band_send_event"] - def test_excludes_contact_tools_without_capability(self): - """Contact tools excluded when CONTACTS capability is absent.""" - from band.core.types import AdapterFeatures - - tools = create_parlant_tools(features=AdapterFeatures()) - tool_names = [t.tool.name for t in tools] + def test_exclude_tools_filters_by_parlant_tool_name(self): + """exclude_tools should remove generated Parlant ToolEntry objects.""" + tools = create_parlant_tools( + features=AdapterFeatures(exclude_tools=frozenset({"band_send_event"})) + ) + tool_names = [entry.tool.name for entry in tools] + assert "band_send_event" not in tool_names assert "band_send_message" in tool_names - assert "band_create_chatroom" in tool_names - assert "band_list_contacts" not in tool_names - assert "band_add_contact" not in tool_names - assert "band_remove_contact" not in tool_names - assert "band_list_contact_requests" not in tool_names - assert "band_respond_contact_request" not in tool_names - - def test_includes_contact_tools_with_capability(self): - """Contact tools included when CONTACTS capability is present.""" - from band.core.types import AdapterFeatures, Capability + def test_include_categories_filters_generated_tools(self): + """Category filters should use canonical contact and memory name sets.""" tools = create_parlant_tools( - features=AdapterFeatures(capabilities={Capability.CONTACTS}) + features=AdapterFeatures( + capabilities=frozenset({Capability.CONTACTS, Capability.MEMORY}), + include_categories=frozenset({"memory"}), + ) ) - tool_names = [t.tool.name for t in tools] + tool_names = [entry.tool.name for entry in tools] - assert "band_list_contacts" in tool_names - assert "band_add_contact" in tool_names - assert "band_remove_contact" in tool_names - assert "band_list_contact_requests" in tool_names - assert "band_respond_contact_request" in tool_names + assert "band_store_memory" in tool_names + assert "band_send_message" not in tool_names + assert "band_list_contacts" not in tool_names - def test_includes_contact_tools_when_no_features(self): - """Contact tools included when features is None (backward compat).""" - tools = create_parlant_tools(features=None) - tool_names = [t.tool.name for t in tools] + def test_includes_additional_custom_tools(self): + """CustomToolDef tools should be exposed as generated Parlant tools.""" + tools = create_parlant_tools(additional_tools=[(CalculatorInput, calculate)]) + calculator = next(entry for entry in tools if entry.tool.name == "calculator") - assert "band_list_contacts" in tool_names - assert "band_respond_contact_request" in tool_names + assert calculator.tool.description + assert list(calculator.tool.parameters) == ["value"] + assert calculator.tool.required == ["value"] class TestParlantToolFunctions: - """Tests for individual Parlant tool functions.""" + """Tests for generated Parlant tool wrapper execution.""" def setup_method(self): - """Clear registry and set up mocks before each test.""" - _session_tools.clear() - _session_message_sent.clear() + """Clear registry before each test.""" + _session_contexts.clear() @pytest.fixture def mock_tools(self): - """Create mock AgentToolsProtocol (MagicMock base, AsyncMock methods).""" + """Create mock AgentToolsProtocol.""" tools = MagicMock() - tools.send_message = AsyncMock() - tools.send_event = AsyncMock() - tools.add_participant = AsyncMock(return_value={"status": "added"}) - tools.remove_participant = AsyncMock() - tools.lookup_peers = AsyncMock( - return_value={ - "peers": [ - {"name": "Agent1", "description": "Test agent", "type": "Agent"} - ], - "metadata": {"page": 1, "total_pages": 1}, - } - ) - tools.get_participants = AsyncMock( - return_value=[{"name": "User1", "type": "User"}] - ) - tools.create_chatroom = AsyncMock(return_value="new-room-123") + tools.execute_tool_call = AsyncMock(return_value={"status": "ok"}) + tools.send_event = AsyncMock(return_value={"status": "sent"}) return tools @pytest.fixture def mock_context(self): - """Create mock ToolContext. - - Uses ``SimpleNamespace`` so that accessing any attribute not - explicitly set raises ``AttributeError`` — this catches tests - that accidentally depend on attributes beyond ``session_id``. - ``MagicMock(spec=ToolContext)`` is not used because ``ToolContext`` - lives in ``parlant.core.tools`` which may not be installed. - """ - from types import SimpleNamespace - + """Create minimal Parlant ToolContext-like object.""" return SimpleNamespace(session_id="test-session-123") @pytest.fixture def parlant_tools(self): - """Create Parlant tools from the real create_parlant_tools.""" - tools = create_parlant_tools() - # Build a dict mapping tool name to the tool's function - return {entry.tool.name: entry.function for entry in tools} + """Create generated Parlant tool functions keyed by tool name.""" + return {entry.tool.name: entry.function for entry in create_parlant_tools()} @pytest.mark.asyncio - async def test_send_message_calls_tools_send_message( + async def test_generated_wrapper_calls_execute_tool_call( self, parlant_tools, mock_tools, mock_context ): - """Should call tools.send_message with parsed mentions.""" + """Generated wrappers should route through the canonical dispatcher.""" set_session_tools(mock_context.session_id, mock_tools) - send_message = parlant_tools["band_send_message"] - result = await send_message(mock_context, "Hello world", "Alice, Bob") + result = await parlant_tools["band_send_event"]( + mock_context, + "Investigating", + "thought", + None, + ) - mock_tools.send_message.assert_called_once_with("Hello world", ["Alice", "Bob"]) - assert "Message sent to Alice, Bob" in result.data + mock_tools.execute_tool_call.assert_awaited_once_with( + "band_send_event", + {"content": "Investigating", "message_type": "thought", "metadata": None}, + ) + assert result.data == '{"status": "ok"}' @pytest.mark.asyncio - async def test_send_message_marks_message_sent( + async def test_generated_wrapper_coerces_json_dict_parameters( self, parlant_tools, mock_tools, mock_context ): - """Should mark message as sent after successful send.""" + """Dict fields exposed as strings should be parsed before validation.""" set_session_tools(mock_context.session_id, mock_tools) - send_message = parlant_tools["band_send_message"] - await send_message(mock_context, "Hello", "Alice") + await parlant_tools["band_send_event"]( + mock_context, + "Investigating", + "thought", + '{"step": 1}', + ) - assert was_message_sent(mock_context.session_id) is True + mock_tools.execute_tool_call.assert_awaited_once_with( + "band_send_event", + { + "content": "Investigating", + "message_type": "thought", + "metadata": {"step": 1}, + }, + ) @pytest.mark.asyncio - async def test_send_message_returns_error_without_tools( - self, parlant_tools, mock_context + async def test_generated_wrapper_rejects_invalid_json_dict_parameters( + self, parlant_tools, mock_tools, mock_context ): - """Should return error when no tools available.""" - send_message = parlant_tools["band_send_message"] - result = await send_message(mock_context, "Hello", "Alice") + """Invalid JSON should be model-visible and not call the platform tool.""" + set_session_tools(mock_context.session_id, mock_tools) + + result = await parlant_tools["band_send_event"]( + mock_context, + "Investigating", + "thought", + "{bad json", + ) - assert "Error: No tools available" in result.data + mock_tools.execute_tool_call.assert_not_awaited() + assert "metadata must be valid JSON" in result.data @pytest.mark.asyncio - async def test_send_message_requires_mentions( + async def test_send_message_marks_sent_after_success( self, parlant_tools, mock_tools, mock_context ): - """Should return error when no mentions provided.""" + """Delivery marker should be set only after canonical send_message succeeds.""" set_session_tools(mock_context.session_id, mock_tools) - send_message = parlant_tools["band_send_message"] - result = await send_message(mock_context, "Hello", "") + result = await parlant_tools["band_send_message"]( + mock_context, + "Hello", + ["@alice"], + ) - assert "At least one mention is required" in result.data + mock_tools.execute_tool_call.assert_awaited_once_with( + "band_send_message", + {"content": "Hello", "mentions": ["@alice"]}, + ) + assert result.data == '{"status": "ok"}' + assert was_message_sent(mock_context.session_id) is True @pytest.mark.asyncio - async def test_send_message_translates_band_tool_error( + async def test_send_message_does_not_mark_sent_after_tool_error( self, parlant_tools, mock_tools, mock_context ): - """BandToolError from underlying tool must surface as ToolResult, not crash. - - Pins the wrapper translation contract: framework wrappers must catch - BandToolError raised by AgentTools and return a model-visible - failure value so the LLM can recover, instead of letting the exception - crash the turn. - """ - from band.core.exceptions import BandToolError - - mock_tools.send_message.side_effect = BandToolError( - "Backend rejected message: 503 Service Unavailable" + """Failed send_message wrapper calls must not count as delivery.""" + mock_tools.execute_tool_call.return_value = ( + "Error executing band_send_message: boom" ) set_session_tools(mock_context.session_id, mock_tools) - send_message = parlant_tools["band_send_message"] - # Must NOT raise — wrapper translates the exception to a tool failure - result = await send_message(mock_context, "Hello", "Alice") + result = await parlant_tools["band_send_message"]( + mock_context, + "Hello", + ["@alice"], + ) - # Result is a ToolResult with the error text visible to the LLM - assert "Error sending message" in result.data - assert "503" in result.data + assert result.data == "Error executing band_send_message: boom" + assert was_message_sent(mock_context.session_id) is False @pytest.mark.asyncio - async def test_send_event_calls_tools_send_event( - self, parlant_tools, mock_tools, mock_context + async def test_tool_returns_error_without_session_tools( + self, parlant_tools, mock_context ): - """Should call tools.send_event with correct parameters.""" - set_session_tools(mock_context.session_id, mock_tools) - - send_event = parlant_tools["band_send_event"] - result = await send_event(mock_context, "Thinking...", "thought") + """Wrapper should return a model-visible error when no session tools exist.""" + result = await parlant_tools["band_send_message"]( + mock_context, + "Hello", + ["@alice"], + ) - mock_tools.send_event.assert_called_once_with("Thinking...", "thought", None) - assert "Event (thought) sent successfully" in result.data + assert result.data == "Error: No tools available in current context" + assert was_message_sent(mock_context.session_id) is False @pytest.mark.asyncio - async def test_send_event_validates_message_type( + async def test_tool_translates_dispatcher_exception( self, parlant_tools, mock_tools, mock_context ): - """Should reject invalid message types.""" + """Unexpected dispatcher exceptions should become model-visible tool errors.""" + mock_tools.execute_tool_call.side_effect = RuntimeError("Connection failed") set_session_tools(mock_context.session_id, mock_tools) - send_event = parlant_tools["band_send_event"] - result = await send_event(mock_context, "Test", "invalid_type") + result = await parlant_tools["band_send_event"]( + mock_context, + "Investigating", + "thought", + None, + ) - assert "Invalid message_type" in result.data + assert "Error executing band_send_event: Connection failed" in result.data @pytest.mark.asyncio - async def test_add_participant_calls_tools( - self, parlant_tools, mock_tools, mock_context - ): - """Should call tools.add_participant.""" - set_session_tools(mock_context.session_id, mock_tools) + async def test_additional_custom_tool_executes_with_validation(self, mock_context): + """Generated custom wrappers should validate through CustomToolDef.""" + tools = { + entry.tool.name: entry.function + for entry in create_parlant_tools( + additional_tools=[(CalculatorInput, calculate)] + ) + } - add_participant = parlant_tools["band_add_participant"] - result = await add_participant(mock_context, "Research Agent") + result = await tools["calculator"](mock_context, 41) - mock_tools.add_participant.assert_called_once_with("Research Agent", "member") - assert "Successfully added 'Research Agent'" in result.data + assert result.data == "42" @pytest.mark.asyncio - async def test_remove_participant_calls_tools( - self, parlant_tools, mock_tools, mock_context - ): - """Should call tools.remove_participant.""" - set_session_tools(mock_context.session_id, mock_tools) + async def test_additional_custom_tool_returns_validation_error(self, mock_context): + """Invalid custom tool args should be model-visible.""" + tools = { + entry.tool.name: entry.function + for entry in create_parlant_tools( + additional_tools=[(CalculatorInput, calculate)] + ) + } - remove_participant = parlant_tools["band_remove_participant"] - result = await remove_participant(mock_context, "Research Agent") + result = await tools["calculator"](mock_context, "not-an-int") - mock_tools.remove_participant.assert_called_once_with("Research Agent") - assert "Successfully removed 'Research Agent'" in result.data + assert "Invalid arguments for calculator" in result.data - @pytest.mark.asyncio - async def test_lookup_peers_returns_formatted_list( - self, parlant_tools, mock_tools, mock_context - ): - """Should return formatted list of peers.""" - set_session_tools(mock_context.session_id, mock_tools) - lookup_peers = parlant_tools["band_lookup_peers"] - result = await lookup_peers(mock_context) +class TestRealTimeExecutionReporting: + """Tool wrappers emit tool_call/tool_result Band events as the tool runs. - # Pagination is hardcoded in the implementation (page=1, page_size=50) - mock_tools.lookup_peers.assert_called_once_with(page=1, page_size=50) - assert "Available agents" in result.data - assert "Agent1" in result.data + Reporting happens inside the wrapper (the real execution point) rather than + by draining Parlant's session log after the turn, so the events are ordered + correctly relative to the actual side effects and emitted exactly once. + """ - @pytest.mark.asyncio - async def test_lookup_peers_handles_empty_result( - self, parlant_tools, mock_tools, mock_context - ): - """Should handle empty peers list.""" - mock_tools.lookup_peers.return_value = {"peers": [], "metadata": {}} - set_session_tools(mock_context.session_id, mock_tools) + def setup_method(self): + _session_contexts.clear() - lookup_peers = parlant_tools["band_lookup_peers"] - result = await lookup_peers(mock_context) + @pytest.fixture + def mock_context(self): + return SimpleNamespace(session_id="session-rt") - assert "No available agents found" in result.data + @pytest.fixture + def builtin_tools(self): + return {entry.tool.name: entry.function for entry in create_parlant_tools()} @pytest.mark.asyncio - async def test_get_participants_returns_formatted_list( - self, parlant_tools, mock_tools, mock_context + async def test_builtin_tool_reports_call_and_result_when_emit_enabled( + self, builtin_tools, mock_context ): - """Should return formatted list of participants.""" - set_session_tools(mock_context.session_id, mock_tools) + """A non-silent builtin tool emits a paired tool_call/tool_result.""" + tools = MagicMock() + tools.execute_tool_call = AsyncMock(return_value={"participants": ["alice"]}) + tools.send_event = AsyncMock(return_value={"status": "sent"}) + set_session_tools(mock_context.session_id, tools, emit_execution=True) - get_participants = parlant_tools["band_get_participants"] - result = await get_participants(mock_context) + await builtin_tools["band_get_participants"](mock_context) - mock_tools.get_participants.assert_called_once() - assert "Current participants" in result.data - assert "User1" in result.data + types = [c.kwargs["message_type"] for c in tools.send_event.await_args_list] + assert types == ["tool_call", "tool_result"] + call_payload = json.loads(tools.send_event.await_args_list[0].kwargs["content"]) + result_payload = json.loads( + tools.send_event.await_args_list[1].kwargs["content"] + ) + assert call_payload["name"] == "band_get_participants" + assert result_payload["name"] == "band_get_participants" + assert result_payload["output"] == {"participants": ["alice"]} + # call and result share one stable id for correlation + assert call_payload["tool_call_id"] == result_payload["tool_call_id"] @pytest.mark.asyncio - async def test_get_participants_handles_empty_room( - self, parlant_tools, mock_tools, mock_context + async def test_no_execution_events_when_emit_disabled( + self, builtin_tools, mock_context ): - """Should handle empty participants list.""" - mock_tools.get_participants.return_value = [] - set_session_tools(mock_context.session_id, mock_tools) + """With emit off the tool still runs but produces no execution events.""" + tools = MagicMock() + tools.execute_tool_call = AsyncMock(return_value={"participants": []}) + tools.send_event = AsyncMock() + set_session_tools(mock_context.session_id, tools, emit_execution=False) - get_participants = parlant_tools["band_get_participants"] - result = await get_participants(mock_context) + await builtin_tools["band_get_participants"](mock_context) - assert "No participants in the room" in result.data + tools.execute_tool_call.assert_awaited_once() + tools.send_event.assert_not_called() @pytest.mark.asyncio - async def test_create_chatroom_calls_tools( - self, parlant_tools, mock_tools, mock_context + async def test_platform_send_tools_are_not_reported( + self, builtin_tools, mock_context ): - """Should call tools.create_chatroom.""" - set_session_tools(mock_context.session_id, mock_tools) + """send_message/send_event already create Band effects, so no echo event.""" + tools = MagicMock() + tools.execute_tool_call = AsyncMock(return_value={"status": "sent"}) + tools.send_event = AsyncMock() + set_session_tools(mock_context.session_id, tools, emit_execution=True) - create_chatroom = parlant_tools["band_create_chatroom"] - result = await create_chatroom(mock_context, "task-456") + await builtin_tools["band_send_message"](mock_context, "Hi", ["@alice"]) + await builtin_tools["band_send_event"](mock_context, "thinking", "thought") - mock_tools.create_chatroom.assert_called_once_with("task-456") - assert "Created new chat room: new-room-123" in result.data + tools.send_event.assert_not_called() + assert tools.execute_tool_call.await_count == 2 @pytest.mark.asyncio - async def test_create_chatroom_handles_empty_task_id( - self, parlant_tools, mock_tools, mock_context + async def test_tool_call_emitted_before_execution( + self, builtin_tools, mock_context ): - """Should handle empty task_id.""" - set_session_tools(mock_context.session_id, mock_tools) + """Ordering oracle: tool_call precedes the side effect, tool_result follows.""" + order: list[str] = [] - create_chatroom = parlant_tools["band_create_chatroom"] - result = await create_chatroom(mock_context, "") + async def fake_execute(name, args): + order.append("execute") + return {"ok": True} - mock_tools.create_chatroom.assert_called_once_with(None) - assert "Created new chat room" in result.data + async def fake_send_event(*, content, message_type): + order.append(message_type) + return {"status": "sent"} - @pytest.mark.asyncio - async def test_tool_handles_exception( - self, parlant_tools, mock_tools, mock_context - ): - """Should return error message when tool raises exception.""" - mock_tools.send_message.side_effect = Exception("Connection failed") - set_session_tools(mock_context.session_id, mock_tools) + tools = MagicMock() + tools.execute_tool_call = AsyncMock(side_effect=fake_execute) + tools.send_event = AsyncMock(side_effect=fake_send_event) + set_session_tools(mock_context.session_id, tools, emit_execution=True) + + await builtin_tools["band_get_participants"](mock_context) - send_message = parlant_tools["band_send_message"] - result = await send_message(mock_context, "Hello", "Alice") + assert order == ["tool_call", "execute", "tool_result"] - assert "Error sending message: Connection failed" in result.data + @pytest.mark.asyncio + async def test_custom_tool_reports_when_emit_enabled(self, mock_context): + """Custom tools route through the same real-time reporting path.""" + tools = MagicMock() + tools.send_event = AsyncMock(return_value={"status": "sent"}) + wrappers = { + entry.tool.name: entry.function + for entry in create_parlant_tools( + additional_tools=[(CalculatorInput, calculate)] + ) + } + set_session_tools(mock_context.session_id, tools, emit_execution=True) + + await wrappers["calculator"](mock_context, 41) + + types = [c.kwargs["message_type"] for c in tools.send_event.await_args_list] + assert types == ["tool_call", "tool_result"] + result_payload = json.loads( + tools.send_event.await_args_list[1].kwargs["content"] + ) + assert result_payload["name"] == "calculator" + assert result_payload["output"] == "42" diff --git a/tests/test_run_agent.py b/tests/test_run_agent.py new file mode 100644 index 000000000..d9ce01eba --- /dev/null +++ b/tests/test_run_agent.py @@ -0,0 +1,280 @@ +"""Tests for the generic example runner.""" + +from pathlib import Path +from types import ModuleType, SimpleNamespace +from unittest.mock import AsyncMock +import importlib.util +import sys + +import pytest + +from band.core.types import Capability, Emit +from band.runtime.types import ContactEventConfig, ContactEventStrategy + + +@pytest.fixture +def run_agent_module(): + """Import examples/run_agent.py as a test module.""" + module_path = str(Path(__file__).resolve().parents[1] / "examples" / "run_agent.py") + spec = importlib.util.spec_from_file_location("example_run_agent", module_path) + assert spec is not None + assert spec.loader is not None + module = importlib.util.module_from_spec(spec) + sys.modules["example_run_agent"] = module + spec.loader.exec_module(module) + return module + + +@pytest.fixture +def fake_parlant_sdk(monkeypatch): + """Install a minimal parlant.sdk module for runner tests.""" + sdk = ModuleType("parlant.sdk") + sdk.NLPServices = SimpleNamespace(openai=object()) + + sdk.server_kwargs = None + + class FakeServer: + def __init__(self, **kwargs): + sdk.server_kwargs = kwargs + self.nlp_service = kwargs["nlp_service"] + self.created_agent = SimpleNamespace(create_guideline=AsyncMock()) + + async def __aenter__(self): + return self + + async def __aexit__(self, exc_type, exc, tb): + return False + + async def create_agent(self, **kwargs): + self.create_agent_kwargs = kwargs + return self.created_agent + + sdk.Server = FakeServer + + package = ModuleType("parlant") + package.sdk = sdk + monkeypatch.setitem(sys.modules, "parlant", package) + monkeypatch.setitem(sys.modules, "parlant.sdk", sdk) + return sdk + + +@pytest.mark.asyncio +async def test_run_parlant_agent_enables_contacts_for_runtime_and_tools( + run_agent_module, + fake_parlant_sdk, + monkeypatch, +): + """Parlant contacts need both Agent.create contact_config and CONTACTS tools.""" + captured = {} + + class FakeAdapter: + def __init__(self, **kwargs): + captured["adapter_kwargs"] = kwargs + + class FakeAgent: + @classmethod + def create(cls, **kwargs): + captured["agent_create_kwargs"] = kwargs + return SimpleNamespace(run=AsyncMock()) + + def fake_create_parlant_tools(features=None, *, legacy_defaults=None): + captured["tool_features"] = features + captured["legacy_defaults"] = legacy_defaults + return ["tool-ref"] + + import band.adapters + import band.integrations.parlant.tools + + monkeypatch.setattr(band.adapters, "ParlantAdapter", FakeAdapter) + monkeypatch.setattr( + band.integrations.parlant.tools, + "create_parlant_tools", + fake_create_parlant_tools, + ) + monkeypatch.setattr(run_agent_module, "Agent", FakeAgent) + + contact_config = ContactEventConfig(strategy=ContactEventStrategy.HUB_ROOM) + + await run_agent_module.run_parlant_agent( + agent_id="agent-id", + api_key="api-key", + rest_url="https://example.test", + ws_url="wss://example.test/socket", + model=None, + custom_section="Be concise", + enable_streaming=False, + contact_config=contact_config, + logger=SimpleNamespace(warning=lambda *args: None, info=lambda *args: None), + ) + + assert fake_parlant_sdk.server_kwargs["port"] == 0 + assert fake_parlant_sdk.server_kwargs["tool_service_port"] == 0 + assert captured["agent_create_kwargs"]["contact_config"] is contact_config + assert captured["adapter_kwargs"]["features"] is captured["tool_features"] + assert captured["tool_features"].capabilities == frozenset({Capability.CONTACTS}) + assert captured["legacy_defaults"] is False + + +@pytest.mark.asyncio +async def test_run_parlant_agent_broadcast_contacts_do_not_enable_contact_tools( + run_agent_module, + fake_parlant_sdk, + monkeypatch, +): + """Broadcast-only contact mode should not grant LLM contact-management tools.""" + captured = {} + + class FakeAdapter: + def __init__(self, **kwargs): + captured["adapter_kwargs"] = kwargs + + class FakeAgent: + @classmethod + def create(cls, **kwargs): + captured["agent_create_kwargs"] = kwargs + return SimpleNamespace(run=AsyncMock()) + + def fake_create_parlant_tools(features=None, *, legacy_defaults=None): + captured["tool_features"] = features + captured["legacy_defaults"] = legacy_defaults + return ["tool-ref"] + + import band.adapters + import band.integrations.parlant.tools + + monkeypatch.setattr(band.adapters, "ParlantAdapter", FakeAdapter) + monkeypatch.setattr( + band.integrations.parlant.tools, + "create_parlant_tools", + fake_create_parlant_tools, + ) + monkeypatch.setattr(run_agent_module, "Agent", FakeAgent) + + contact_config = ContactEventConfig( + strategy=ContactEventStrategy.DISABLED, + broadcast_changes=True, + ) + + await run_agent_module.run_parlant_agent( + agent_id="agent-id", + api_key="api-key", + rest_url="https://example.test", + ws_url="wss://example.test/socket", + model=None, + custom_section="Be concise", + enable_streaming=False, + contact_config=contact_config, + logger=SimpleNamespace(warning=lambda *args: None, info=lambda *args: None), + ) + + assert captured["agent_create_kwargs"]["contact_config"] is contact_config + assert captured["adapter_kwargs"]["features"] is None + assert captured["tool_features"] is None + assert captured["legacy_defaults"] is False + + +@pytest.mark.asyncio +async def test_run_parlant_agent_excludes_contact_tools_without_contacts( + run_agent_module, + fake_parlant_sdk, + monkeypatch, +): + """Without contact_config the runner should not grant contact-management tools.""" + captured = {} + + class FakeAdapter: + def __init__(self, **kwargs): + captured["adapter_kwargs"] = kwargs + + class FakeAgent: + @classmethod + def create(cls, **kwargs): + captured["agent_create_kwargs"] = kwargs + return SimpleNamespace(run=AsyncMock()) + + def fake_create_parlant_tools(features=None, *, legacy_defaults=None): + captured["tool_features"] = features + captured["legacy_defaults"] = legacy_defaults + return ["tool-ref"] + + import band.adapters + import band.integrations.parlant.tools + + monkeypatch.setattr(band.adapters, "ParlantAdapter", FakeAdapter) + monkeypatch.setattr( + band.integrations.parlant.tools, + "create_parlant_tools", + fake_create_parlant_tools, + ) + monkeypatch.setattr(run_agent_module, "Agent", FakeAgent) + + await run_agent_module.run_parlant_agent( + agent_id="agent-id", + api_key="api-key", + rest_url="https://example.test", + ws_url="wss://example.test/socket", + model=None, + custom_section="Be concise", + enable_streaming=False, + contact_config=None, + logger=SimpleNamespace(warning=lambda *args: None, info=lambda *args: None), + ) + + assert captured["agent_create_kwargs"].get("contact_config") is None + assert captured["adapter_kwargs"]["features"] is None + assert captured["tool_features"] is None + assert captured["legacy_defaults"] is False + + +@pytest.mark.asyncio +async def test_run_parlant_agent_enables_execution_reporting( + run_agent_module, + fake_parlant_sdk, + monkeypatch, +): + """--streaming should enable Emit.EXECUTION for the Parlant adapter.""" + captured = {} + + class FakeAdapter: + def __init__(self, **kwargs): + captured["adapter_kwargs"] = kwargs + + class FakeAgent: + @classmethod + def create(cls, **kwargs): + captured["agent_create_kwargs"] = kwargs + return SimpleNamespace(run=AsyncMock()) + + def fake_create_parlant_tools(features=None, *, legacy_defaults=None): + captured["tool_features"] = features + captured["legacy_defaults"] = legacy_defaults + return ["tool-ref"] + + import band.adapters + import band.integrations.parlant.tools + + monkeypatch.setattr(band.adapters, "ParlantAdapter", FakeAdapter) + monkeypatch.setattr( + band.integrations.parlant.tools, + "create_parlant_tools", + fake_create_parlant_tools, + ) + monkeypatch.setattr(run_agent_module, "Agent", FakeAgent) + + await run_agent_module.run_parlant_agent( + agent_id="agent-id", + api_key="api-key", + rest_url="https://example.test", + ws_url="wss://example.test/socket", + model=None, + custom_section="Be concise", + enable_streaming=True, + contact_config=None, + logger=SimpleNamespace(warning=lambda *args: None, info=lambda *args: None), + ) + + features = captured["adapter_kwargs"]["features"] + assert captured["tool_features"] is features + assert features.emit == frozenset({Emit.EXECUTION}) + assert features.capabilities == frozenset() + assert captured["legacy_defaults"] is False