diff --git a/python/samples/04-hosting/a2a/README.md b/python/samples/04-hosting/a2a/README.md index 2ede8b8a3d..0affc84e19 100644 --- a/python/samples/04-hosting/a2a/README.md +++ b/python/samples/04-hosting/a2a/README.md @@ -1,34 +1,57 @@ # A2A Agent Examples -This folder contains examples demonstrating how to create and use agents with the A2A (Agent2Agent) protocol from the `agent_framework` package to communicate with remote A2A agents. +This sample demonstrates how to host and consume agents using the [A2A (Agent2Agent) protocol](https://a2a-protocol.org/latest/) with the `agent_framework` package. There are two runnable entry points: -By default the A2AAgent waits for the remote agent to finish before returning (`background=False`), so long-running A2A tasks are handled transparently. For advanced scenarios where you need to poll or resubscribe to in-progress tasks using continuation tokens, see the [background responses sample](../../02-agents/background_responses.py). +| Run this file | To... | +|---------------|-------| +| **[`a2a_server.py`](a2a_server.py)** | Host an Agent Framework agent as an A2A-compliant server. | +| **[`agent_with_a2a.py`](agent_with_a2a.py)** | Connect to an A2A server and send requests (non-streaming and streaming). | -For more information about the A2A protocol specification, visit: https://a2a-protocol.org/latest/ - -## Examples +The remaining files are supporting modules used by the server: | File | Description | |------|-------------| -| [`agent_with_a2a.py`](agent_with_a2a.py) | Demonstrates agent discovery, non-streaming and streaming responses using the A2A protocol. | +| [`agent_definitions.py`](agent_definitions.py) | Agent and AgentCard factory definitions for invoice, policy, and logistics agents. | +| [`agent_executor.py`](agent_executor.py) | Bridges the a2a-sdk `AgentExecutor` interface to Agent Framework agents. | +| [`invoice_data.py`](invoice_data.py) | Mock invoice data and tool functions for the invoice agent. | +| [`a2a_server.http`](a2a_server.http) | REST Client requests for testing the server directly from VS Code. | ## Environment Variables -Make sure to set the following environment variables before running the example: +Make sure to set the following environment variables before running the examples: + +### Required (Server) +- `AZURE_AI_PROJECT_ENDPOINT` — Your Azure AI Foundry project endpoint +- `AZURE_OPENAI_RESPONSES_DEPLOYMENT_NAME` — Model deployment name (e.g. `gpt-4o`) + +### Required (Client) +- `A2A_AGENT_HOST` — URL of the A2A server (e.g. `http://localhost:5001/`) -### Required -- `A2A_AGENT_HOST`: URL of a single A2A agent (for simple sample, e.g., `http://localhost:5001/`) +## Quick Start +All commands below should be run from this directory: + +```powershell +cd python/samples/04-hosting/a2a +``` + +### 1. Start the A2A Server + +Pick an agent type and start the server (each in its own terminal): + +```powershell +uv run python a2a_server.py --agent-type invoice --port 5000 +uv run python a2a_server.py --agent-type policy --port 5001 +uv run python a2a_server.py --agent-type logistics --port 5002 +``` -## Quick Testing with .NET A2A Servers +You can run one agent or all three — each listens on its own port. -For quick testing and demonstration, you can use the pre-built .NET A2A servers from this repository: +### 2. Run the A2A Client -**Quick Testing Reference**: Use the .NET A2A Client Server sample at: -`..\agent-framework\dotnet\samples\05-end-to-end\A2AClientServer` +In a separate terminal (from the same directory), point the client at a running server: -### Run Python A2A Sample ```powershell -# Simple A2A sample (single agent) +$env:A2A_AGENT_HOST = "http://localhost:5001/" uv run python agent_with_a2a.py ``` diff --git a/python/samples/04-hosting/a2a/a2a_server.http b/python/samples/04-hosting/a2a/a2a_server.http new file mode 100644 index 0000000000..65ff918a1d --- /dev/null +++ b/python/samples/04-hosting/a2a/a2a_server.http @@ -0,0 +1,82 @@ +### Each A2A agent is available at a different host address +@hostInvoice = http://localhost:5000 +@hostPolicy = http://localhost:5001 +@hostLogistics = http://localhost:5002 + +### Query agent card for the invoice agent +GET {{hostInvoice}}/.well-known/agent.json + +### Send a message to the invoice agent +POST {{hostInvoice}} +Content-Type: application/json + +{ + "id": "1", + "jsonrpc": "2.0", + "method": "message/send", + "params": { + "message": { + "kind": "message", + "role": "user", + "messageId": "msg_1", + "parts": [ + { + "kind": "text", + "text": "Show me all invoices for Contoso" + } + ] + } + } +} + +### Query agent card for the policy agent +GET {{hostPolicy}}/.well-known/agent.json + +### Send a message to the policy agent +POST {{hostPolicy}} +Content-Type: application/json + +{ + "id": "2", + "jsonrpc": "2.0", + "method": "message/send", + "params": { + "message": { + "kind": "message", + "role": "user", + "messageId": "msg_2", + "parts": [ + { + "kind": "text", + "text": "What is the policy for short shipments?" + } + ] + } + } +} + +### Query agent card for the logistics agent +GET {{hostLogistics}}/.well-known/agent.json + +### Send a message to the logistics agent +POST {{hostLogistics}} +Content-Type: application/json + +{ + "id": "3", + "jsonrpc": "2.0", + "method": "message/send", + "params": { + "message": { + "kind": "message", + "role": "user", + "messageId": "msg_3", + "parts": [ + { + "kind": "text", + "text": "What is the status for SHPMT-SAP-001?" + } + ] + } + } +} diff --git a/python/samples/04-hosting/a2a/a2a_server.py b/python/samples/04-hosting/a2a/a2a_server.py new file mode 100644 index 0000000000..d797bef95d --- /dev/null +++ b/python/samples/04-hosting/a2a/a2a_server.py @@ -0,0 +1,120 @@ +# Copyright (c) Microsoft. All rights reserved. + +import argparse +import os +import sys + +import uvicorn +from a2a.server.apps.jsonrpc.starlette_app import A2AStarletteApplication +from a2a.server.request_handlers.default_request_handler import DefaultRequestHandler +from a2a.server.tasks.inmemory_task_store import InMemoryTaskStore +from agent_definitions import AGENT_CARD_FACTORIES, AGENT_FACTORIES +from agent_executor import AgentFrameworkExecutor +from agent_framework.azure import AzureOpenAIResponsesClient +from azure.identity import AzureCliCredential +from dotenv import load_dotenv + +# Load environment variables from .env file +load_dotenv() + +""" +A2A Server Sample — Host an Agent Framework agent as an A2A endpoint + +This sample creates a Python-based A2A-compliant server that wraps an Agent +Framework agent. The server uses the a2a-sdk's Starlette application to handle +JSON-RPC requests and serves the AgentCard at /.well-known/agent.json. + +Three agent types are available: + - invoice — Answers invoice queries using mock data and function tools. + - policy — Returns a fixed policy response. + - logistics — Returns a fixed logistics response. + +Usage: + uv run python a2a_server.py --agent-type policy --port 5001 + uv run python a2a_server.py --agent-type invoice --port 5000 + uv run python a2a_server.py --agent-type logistics --port 5002 + +Environment variables: + AZURE_AI_PROJECT_ENDPOINT — Your Azure AI Foundry project endpoint + AZURE_OPENAI_RESPONSES_DEPLOYMENT_NAME — Model deployment name (e.g. gpt-4o) +""" + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser(description="A2A Agent Server") + parser.add_argument( + "--agent-type", + choices=["invoice", "policy", "logistics"], + default="policy", + help="Type of agent to host (default: policy)", + ) + parser.add_argument( + "--host", + default="localhost", + help="Host to bind to (default: localhost)", + ) + parser.add_argument( + "--port", + type=int, + default=5001, + help="Port to listen on (default: 5001)", + ) + return parser.parse_args() + + +def main() -> None: + args = parse_args() + + # Validate environment + project_endpoint = os.getenv("AZURE_AI_PROJECT_ENDPOINT") + deployment_name = os.getenv("AZURE_OPENAI_RESPONSES_DEPLOYMENT_NAME") + + if not project_endpoint: + print("Error: AZURE_AI_PROJECT_ENDPOINT environment variable is not set.") + sys.exit(1) + if not deployment_name: + print("Error: AZURE_OPENAI_RESPONSES_DEPLOYMENT_NAME environment variable is not set.") + sys.exit(1) + + # Create the LLM client + credential = AzureCliCredential() + client = AzureOpenAIResponsesClient( + project_endpoint=project_endpoint, + deployment_name=deployment_name, + credential=credential, + ) + + # Create the Agent Framework agent for the chosen type + agent_factory = AGENT_FACTORIES[args.agent_type] + agent = agent_factory(client) + + # Build the A2A server components + url = f"http://{args.host}:{args.port}/" + agent_card = AGENT_CARD_FACTORIES[args.agent_type](url) + executor = AgentFrameworkExecutor(agent) + task_store = InMemoryTaskStore() + request_handler = DefaultRequestHandler( + agent_executor=executor, + task_store=task_store, + ) + + a2a_app = A2AStarletteApplication( + agent_card=agent_card, + http_handler=request_handler, + ) + + print(f"Starting A2A server: {agent_card.name}") + print(f" Agent type : {args.agent_type}") + print(f" Listening : {url}") + print(f" Agent card : {url}.well-known/agent.json") + print() + + uvicorn.run( + a2a_app.build(), + host=args.host, + port=args.port, + ) + + +if __name__ == "__main__": + main() diff --git a/python/samples/04-hosting/a2a/agent_definitions.py b/python/samples/04-hosting/a2a/agent_definitions.py new file mode 100644 index 0000000000..b0e87e485f --- /dev/null +++ b/python/samples/04-hosting/a2a/agent_definitions.py @@ -0,0 +1,169 @@ +# Copyright (c) Microsoft. All rights reserved. + +"""Agent definitions and AgentCard factories for the A2A server sample. + +Provides factory functions to create Agent Framework agents and A2A +AgentCards for the invoice, policy, and logistics agent types. +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +from a2a.types import AgentCapabilities, AgentCard, AgentSkill +from invoice_data import query_by_invoice_id, query_by_transaction_id, query_invoices + +if TYPE_CHECKING: + from agent_framework import Agent + from agent_framework.azure import AzureOpenAIResponsesClient + + +# --------------------------------------------------------------------------- +# Agent instructions +# --------------------------------------------------------------------------- + +INVOICE_INSTRUCTIONS = "You specialize in handling queries related to invoices." + +POLICY_INSTRUCTIONS = """\ +You specialize in handling queries related to policies and customer communications. + +Always reply with exactly this text: + +Policy: Short Shipment Dispute Handling Policy V2.1 + +Summary: "For short shipments reported by customers, first verify internal shipment records +(SAP) and physical logistics scan data (BigQuery). If discrepancy is confirmed and logistics data +shows fewer items packed than invoiced, issue a credit for the missing items. Document the +resolution in SAP CRM and notify the customer via email within 2 business days, referencing the +original invoice and the credit memo number. Use the 'Formal Credit Notification' email +template." +""" + +LOGISTICS_INSTRUCTIONS = """\ +You specialize in handling queries related to logistics. + +Always reply with exactly: + +Shipment number: SHPMT-SAP-001 +Item: TSHIRT-RED-L +Quantity: 900 +""" + +# --------------------------------------------------------------------------- +# Agent factories +# --------------------------------------------------------------------------- + + +def create_invoice_agent(client: AzureOpenAIResponsesClient) -> Agent: + """Create an invoice agent backed by the given client with query tools.""" + return client.as_agent( + name="InvoiceAgent", + instructions=INVOICE_INSTRUCTIONS, + tools=[query_invoices, query_by_transaction_id, query_by_invoice_id], + ) + + +def create_policy_agent(client: AzureOpenAIResponsesClient) -> Agent: + """Create a policy agent backed by the given client.""" + return client.as_agent( + name="PolicyAgent", + instructions=POLICY_INSTRUCTIONS, + ) + + +def create_logistics_agent(client: AzureOpenAIResponsesClient) -> Agent: + """Create a logistics agent backed by the given client.""" + return client.as_agent( + name="LogisticsAgent", + instructions=LOGISTICS_INSTRUCTIONS, + ) + + +# --------------------------------------------------------------------------- +# AgentCard factories +# --------------------------------------------------------------------------- + +_CAPABILITIES = AgentCapabilities(streaming=True, push_notifications=False) + + +def get_invoice_agent_card(url: str) -> AgentCard: + """Return an A2A AgentCard for the invoice agent.""" + return AgentCard( + name="InvoiceAgent", + description="Handles requests relating to invoices.", + url=url, + version="1.0.0", + default_input_modes=["text"], + default_output_modes=["text"], + capabilities=_CAPABILITIES, + skills=[ + AgentSkill( + id="id_invoice_agent", + name="InvoiceQuery", + description="Handles requests relating to invoices.", + tags=["invoice", "agent-framework"], + examples=["List the latest invoices for Contoso."], + ), + ], + ) + + +def get_policy_agent_card(url: str) -> AgentCard: + """Return an A2A AgentCard for the policy agent.""" + return AgentCard( + name="PolicyAgent", + description="Handles requests relating to policies and customer communications.", + url=url, + version="1.0.0", + default_input_modes=["text"], + default_output_modes=["text"], + capabilities=_CAPABILITIES, + skills=[ + AgentSkill( + id="id_policy_agent", + name="PolicyAgent", + description="Handles requests relating to policies and customer communications.", + tags=["policy", "agent-framework"], + examples=["What is the policy for short shipments?"], + ), + ], + ) + + +def get_logistics_agent_card(url: str) -> AgentCard: + """Return an A2A AgentCard for the logistics agent.""" + return AgentCard( + name="LogisticsAgent", + description="Handles requests relating to logistics.", + url=url, + version="1.0.0", + default_input_modes=["text"], + default_output_modes=["text"], + capabilities=_CAPABILITIES, + skills=[ + AgentSkill( + id="id_logistics_agent", + name="LogisticsQuery", + description="Handles requests relating to logistics.", + tags=["logistics", "agent-framework"], + examples=["What is the status for SHPMT-SAP-001"], + ), + ], + ) + + +# --------------------------------------------------------------------------- +# Lookup helpers +# --------------------------------------------------------------------------- + +AGENT_FACTORIES = { + "invoice": create_invoice_agent, + "policy": create_policy_agent, + "logistics": create_logistics_agent, +} + +AGENT_CARD_FACTORIES = { + "invoice": get_invoice_agent_card, + "policy": get_policy_agent_card, + "logistics": get_logistics_agent_card, +} diff --git a/python/samples/04-hosting/a2a/agent_executor.py b/python/samples/04-hosting/a2a/agent_executor.py new file mode 100644 index 0000000000..b940be18f8 --- /dev/null +++ b/python/samples/04-hosting/a2a/agent_executor.py @@ -0,0 +1,123 @@ +# Copyright (c) Microsoft. All rights reserved. + +"""AgentExecutor bridge between the a2a-sdk server and Agent Framework agents. + +Implements the a2a-sdk ``AgentExecutor`` interface so that incoming A2A +requests are forwarded to an Agent Framework agent and the response is +published back through the a2a-sdk event queue. +""" + +from __future__ import annotations + +import asyncio +import uuid +from typing import TYPE_CHECKING + +from a2a.server.agent_execution.agent_executor import AgentExecutor +from a2a.types import ( + Message, + Part, + Role, + TaskState, + TaskStatus, + TaskStatusUpdateEvent, + TextPart, +) + +if TYPE_CHECKING: + from a2a.server.agent_execution.context import RequestContext + from a2a.server.events.event_queue import EventQueue + from agent_framework import Agent + + +class AgentFrameworkExecutor(AgentExecutor): + """Bridges A2A protocol requests to an Agent Framework agent. + + For each incoming ``execute`` call the executor: + 1. Extracts the user's text from the A2A ``RequestContext``. + 2. Runs the Agent Framework agent (non-streaming). + 3. Publishes the result as an A2A ``Message`` to the ``EventQueue``. + """ + + def __init__(self, agent: Agent) -> None: + self.agent = agent + + async def execute(self, context: RequestContext, event_queue: EventQueue) -> None: + """Run the agent and publish the response.""" + user_text = context.get_user_input() + if not user_text: + user_text = "Hello" + + task_id = context.task_id or str(uuid.uuid4()) + context_id = context.context_id or str(uuid.uuid4()) + + # Signal that the agent is working + await event_queue.enqueue_event( + TaskStatusUpdateEvent( + task_id=task_id, + context_id=context_id, + status=TaskStatus(state=TaskState.working), + final=False, + ) + ) + + try: + response = await self.agent.run(user_text) + + # Build response text from agent messages + response_parts: list[Part] = [] + for msg in response.messages: + if msg.text: + response_parts.append(TextPart(text=msg.text)) + + if not response_parts: + response_parts.append(TextPart(text=str(response))) + + # Publish the agent's response as a completed message + await event_queue.enqueue_event( + TaskStatusUpdateEvent( + task_id=task_id, + context_id=context_id, + status=TaskStatus( + state=TaskState.completed, + message=Message( + message_id=str(uuid.uuid4()), + role=Role.agent, + parts=response_parts, + ), + ), + final=True, + ) + ) + except asyncio.CancelledError: + raise + except Exception as e: + await event_queue.enqueue_event( + TaskStatusUpdateEvent( + task_id=task_id, + context_id=context_id, + status=TaskStatus( + state=TaskState.failed, + message=Message( + message_id=str(uuid.uuid4()), + role=Role.agent, + parts=[TextPart(text=f"Agent error: {e}")], + ), + ), + final=True, + ) + ) + + async def cancel(self, context: RequestContext, event_queue: EventQueue) -> None: + """Handle cancellation by publishing a canceled status.""" + task_id = context.task_id or str(uuid.uuid4()) + context_id = context.context_id or str(uuid.uuid4()) + + await event_queue.enqueue_event( + TaskStatusUpdateEvent( + task_id=task_id, + context_id=context_id, + status=TaskStatus(state=TaskState.canceled), + final=True, + ) + ) diff --git a/python/samples/04-hosting/a2a/agent_with_a2a.py b/python/samples/04-hosting/a2a/agent_with_a2a.py index 89d43e4b0a..58415b038c 100644 --- a/python/samples/04-hosting/a2a/agent_with_a2a.py +++ b/python/samples/04-hosting/a2a/agent_with_a2a.py @@ -78,16 +78,16 @@ async def main(): # Updates arrive as Server-Sent Events, letting you observe # progress in real time as the remote agent works. print("\n--- Streaming response ---") - async with agent.run("Tell me about yourself", stream=True) as stream: - async for update in stream: - for content in update.contents: - if content.text: - print(f" {content.text}") - - response = await stream.get_final_response() - print(f"\nFinal response ({len(response.messages)} message(s)):") - for message in response.messages: - print(f" {message.text}") + stream = agent.run("Tell me about yourself", stream=True) + async for update in stream: + for content in update.contents: + if content.text: + print(f" {content.text}") + + response = await stream.get_final_response() + print(f"\nFinal response ({len(response.messages)} message(s)):") + for message in response.messages: + print(f" {message.text}") if __name__ == "__main__": diff --git a/python/samples/04-hosting/a2a/invoice_data.py b/python/samples/04-hosting/a2a/invoice_data.py new file mode 100644 index 0000000000..877a00b4d2 --- /dev/null +++ b/python/samples/04-hosting/a2a/invoice_data.py @@ -0,0 +1,167 @@ +# Copyright (c) Microsoft. All rights reserved. + +"""Mock invoice data and tool functions for the A2A server sample. + +Provides mock invoice data and query tools for the A2A server sample, +enabling invoice-related queries through the A2A protocol. +""" + +import json +import random +from dataclasses import dataclass, field +from datetime import datetime, timedelta, timezone +from typing import Annotated + +from agent_framework import tool +from pydantic import Field + + +@dataclass +class Product: + """A product line item on an invoice.""" + + name: str + quantity: int + price_per_unit: float + + @property + def total_price(self) -> float: + return self.quantity * self.price_per_unit + + def to_dict(self) -> dict: + return { + "name": self.name, + "quantity": self.quantity, + "price_per_unit": self.price_per_unit, + "total_price": self.total_price, + } + + +@dataclass +class Invoice: + """An invoice record with products.""" + + transaction_id: str + invoice_id: str + company_name: str + invoice_date: datetime + products: list[Product] = field(default_factory=list) + + @property + def total_invoice_price(self) -> float: + return sum(p.total_price for p in self.products) + + def to_dict(self) -> dict: + return { + "transaction_id": self.transaction_id, + "invoice_id": self.invoice_id, + "company_name": self.company_name, + "invoice_date": self.invoice_date.strftime("%Y-%m-%d"), + "products": [p.to_dict() for p in self.products], + "total_invoice_price": self.total_invoice_price, + } + + +def _random_date_within_last_two_months() -> datetime: + end_date = datetime.now(timezone.utc) + start_date = end_date - timedelta(days=60) + random_days = random.randint(0, 60) + return start_date + timedelta(days=random_days) + + +def _build_invoices() -> list[Invoice]: + """Build 10 mock invoices.""" + return [ + Invoice("TICKET-XYZ987", "INV789", "Contoso", _random_date_within_last_two_months(), [ + Product("T-Shirts", 150, 10.00), + Product("Hats", 200, 15.00), + Product("Glasses", 300, 5.00), + ]), + Invoice("TICKET-XYZ111", "INV111", "XStore", _random_date_within_last_two_months(), [ + Product("T-Shirts", 2500, 12.00), + Product("Hats", 1500, 8.00), + Product("Glasses", 200, 20.00), + ]), + Invoice("TICKET-XYZ222", "INV222", "Cymbal Direct", _random_date_within_last_two_months(), [ + Product("T-Shirts", 1200, 14.00), + Product("Hats", 800, 7.00), + Product("Glasses", 500, 25.00), + ]), + Invoice("TICKET-XYZ333", "INV333", "Contoso", _random_date_within_last_two_months(), [ + Product("T-Shirts", 400, 11.00), + Product("Hats", 600, 15.00), + Product("Glasses", 700, 5.00), + ]), + Invoice("TICKET-XYZ444", "INV444", "XStore", _random_date_within_last_two_months(), [ + Product("T-Shirts", 800, 10.00), + Product("Hats", 500, 18.00), + Product("Glasses", 300, 22.00), + ]), + Invoice("TICKET-XYZ555", "INV555", "Cymbal Direct", _random_date_within_last_two_months(), [ + Product("T-Shirts", 1100, 9.00), + Product("Hats", 900, 12.00), + Product("Glasses", 1200, 15.00), + ]), + Invoice("TICKET-XYZ666", "INV666", "Contoso", _random_date_within_last_two_months(), [ + Product("T-Shirts", 2500, 8.00), + Product("Hats", 1200, 10.00), + Product("Glasses", 1000, 6.00), + ]), + Invoice("TICKET-XYZ777", "INV777", "XStore", _random_date_within_last_two_months(), [ + Product("T-Shirts", 1900, 13.00), + Product("Hats", 1300, 16.00), + Product("Glasses", 800, 19.00), + ]), + Invoice("TICKET-XYZ888", "INV888", "Cymbal Direct", _random_date_within_last_two_months(), [ + Product("T-Shirts", 2200, 11.00), + Product("Hats", 1700, 8.50), + Product("Glasses", 600, 21.00), + ]), + Invoice("TICKET-XYZ999", "INV999", "Contoso", _random_date_within_last_two_months(), [ + Product("T-Shirts", 1400, 10.50), + Product("Hats", 1100, 9.00), + Product("Glasses", 950, 12.00), + ]), + ] + + +# Module-level singleton so dates are stable for the lifetime of the server +INVOICES = _build_invoices() + + +@tool(approval_mode="never_require") +def query_invoices( + company_name: Annotated[str, Field(description="The company name to filter invoices by.")], + start_date: Annotated[str | None, Field(description="Optional start date (YYYY-MM-DD) to filter invoices.")] = None, + end_date: Annotated[str | None, Field(description="Optional end date (YYYY-MM-DD) to filter invoices.")] = None, +) -> str: + """Retrieves invoices for the specified company and optionally within the specified time range.""" + results = [i for i in INVOICES if i.company_name.lower() == company_name.lower()] + + if start_date: + start = datetime.strptime(start_date, "%Y-%m-%d").replace(tzinfo=timezone.utc) + results = [i for i in results if i.invoice_date >= start] + + if end_date: + end = datetime.strptime(end_date, "%Y-%m-%d").replace(tzinfo=timezone.utc) + timedelta(days=1) + results = [i for i in results if i.invoice_date < end] + + return json.dumps([i.to_dict() for i in results], indent=2) + + +@tool(approval_mode="never_require") +def query_by_transaction_id( + transaction_id: Annotated[str, Field(description="The transaction ID to look up (e.g. TICKET-XYZ987).")], +) -> str: + """Retrieves invoice using the transaction id.""" + results = [i for i in INVOICES if i.transaction_id.lower() == transaction_id.lower()] + return json.dumps([i.to_dict() for i in results], indent=2) + + +@tool(approval_mode="never_require") +def query_by_invoice_id( + invoice_id: Annotated[str, Field(description="The invoice ID to look up (e.g. INV789).")], +) -> str: + """Retrieves invoice using the invoice id.""" + results = [i for i in INVOICES if i.invoice_id.lower() == invoice_id.lower()] + return json.dumps([i.to_dict() for i in results], indent=2)