Single agent
Building a single agent
This guide covers building agents using BridgeBaseAgent for sequential, single-pipeline use cases. Use BridgeBaseAgent when your agent:
- Follows a linear, sequential flow.
- Has a single execution path.
- Doesn't need conditional routing between multiple sub-agents.
from bridge_agent_sdk import BridgeBaseAgent, AgentConfig
class MyAgent(BridgeBaseAgent):
"""Your custom agent."""
CONFIG = AgentConfig(
name="my-agent",
version="1.0.0",
description="My agent description",
)
def setup_tools(self) -> list:
"""Return list of LangChain tools available to the agent."""
return []
def setup_prompt(self) -> str:
"""Return the system prompt for the agent."""
return "You are a helpful assistant."Lifecycle methods
Method | Purpose | When Called |
|---|---|---|
setup_tools() | Define available tools | On initialization |
setup_prompt() | Define system prompt | On initialization |
execute() | Main agent logic | On each invocation |
initialize() | Async init (LLM, resources) | Before first execution |
aclose() | Cleanup resources | After execution |
Add single agent
Step 1: Define Your Tools
Create tools in src/tools/my_tools.py:
from langchain_core.tools import tool
from typing import List, Dict, Any
@tooldef search_database(query: str) -> List[Dict[str, Any]]:Search the database for records matching the query.
query: The search query string
Returns: List of matching records
return [{"id": 1, "name": "Result 1"}]
@tool
def get_metrics(metric_name: str, time_range: str = "1h") -> Dict[str, Any]:Retrieve metrics for the specified metric name.
Args:
metric_name: Name of the metric to retrieve.
time_range: Time range (e.g., '1h', '24h', '7d')
Returns: Metric data with timestamps and values
return {"metric": metric_name, "values": [1, 2, 3]}Step 2: Implement Your Agent
Create agent in src/agents/my_agent.py:
Import logging
from typing import Dict, Any, Union, Optional
from bridge_agent_sdk import BridgeBaseAgent, AgentConfig, AgentInput
from bridge_agent_sdk.execution_context import ExecutionContext
from src.tools.my_tools import search_database, get_metrics
logger = logging.getLogger(__name__)
class MyAgent(BridgeBaseAgent):Search agent data and provide insights
This agent:
1. Receives a user query
2. Searches the database for relevant information
3. Analyzes the results using LLM
4. Returns formatted insights
CONFIG = AgentConfig(
name="my-agent",
version="1.0.0",
description="Data analysis agent",
)
def setup_tools(self) -> list:Configure tools available to this agent.
return [search_database, get_metrics]
def setup_prompt(self) -> str:Configure the system prompt for this agent.
Your capabilities:
- Search databases for relevant information
- Retrieve and analyze metrics
Guidelines:
- Always search for data before making conclusions
- Provide specific, actionable insights
- Be concise but comprehensive
async def execute(
self,
input_data: Union[Dict[str, Any], AgentInput],
context: Optional[ExecutionContext] = None,
) -> Dict[str, Any]:
"""Execute the agent logic.
Args:
input_data: AgentInput or dict with input data
context: Optional ExecutionContext from the platform
Returns:
Dict with agent output
"""
if isinstance(input_data, dict):
agent_input = AgentInput(**input_data)
else:
agent_input = input_data
logger.info(f"Executing MyAgent with content: {str(agent_input.content)[:50]}")
try:
# Create LLM using base class helper
state = agent_input.context or {}
llm = self._create_llm(state)
messages = [
{"role": "system", "content": self.setup_prompt()},
{"role": "user", "content": str(agent_input.content)},
]
response = llm.invoke(messages)
return {
"content": response.content,
"status": "success",
}
except Exception as e:
logger.error(f"Agent execution failed: {{e}}")
return {
"content": "",
"status": "error",
"error": str(e),
}Step 3: Create the entrypoint
Create main.py:
"""Application entrypoint."""
import logging
from dotenv import load_dotenv
load_dotenv(override=False)
from bridge_agent_sdk import run_agent
from src.agents.my_agent import MyAgent
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)
AGENTS = {"my_agent": MyAgent}
AGENT_NAME_MAP = {"my_agent": "my_agent"}
if __name__ == "__main__":
run_agent(
AGENTS,
agent_name_map=AGENT_NAME_MAP,
description="My Agent Runner",
)Adding MCP tools
Your agent can use tools from MCP servers:
import os
from bridge_agent_sdk import MCPClient
class MyAgent(BridgeBaseAgent):
CONFIG = AgentConfig(
name="my-agent", version="1.0.0", description="MCP agent"
)
def setup_tools(self):
return []
def setup_prompt(self):
return "You are a helpful assistant."
async def execute(self, input_data, context=None):
if isinstance(input_data, dict):
agent_input = AgentInput(**input_data)
else:
agent_input = input_data
platform_context = (agent_input.context or {}).get("platform_context", {})
mcp = MCPClient(
dev_mode=(os.getenv("KAIF_MODE") != "production"),
agent_payload=platform_context if os.getenv("KAIF_MODE") != "local_dev" else None,
)
# Discover available tools
tools = await mcp.discover_tools()
# Execute a query — use call_tool_parsed for bridge_execute_query
success, parsed = await mcp.call_tool_parsed(
"bridge_execute_query",
{"query": 'SELECT * FROM "ITSM".incident LIMIT 10'},
)
rows = parsed.get("data", []) if success else []
return {"content": str(rows), "status": "success" if success else "error"}Testing your agent
Create tests in tests/test_my_agent.py:
"""Tests for MyAgent."""
import pytest
from unittest.mock import patch, MagicMock
from bridge_agent_sdk import AgentInput
from bridge_agent_sdk.testing import create_test_execution_context
from src.agents.my_agent import MyAgent
@pytest.fixture
def agent():
"""Create agent instance for testing."""
return MyAgent()
@pytest.fixture
def agent_input():
"""Create test AgentInput."""
return AgentInput(
content="Show me recent incidents",
metadata={"thread_id": "test-session-123"},
context={"agent_payload": {"account_id": "test-account"}},
)
@pytest.mark.asyncio
async def test_agent_execution_success(agent, agent_input):
"""Test successful agent execution."""
mock_response = MagicMock()
mock_response.content = "Here are the recent incidents..."
with patch.object(agent, '_create_llm') as mock_create_llm:
mock_llm = MagicMock()
mock_llm.invoke.return_value = mock_response
mock_create_llm.return_value = mock_llm
result = await agent.execute(agent_input)
assert result["status"] == "success"
assert result["content"] is not None
@pytest.mark.asyncio
async def test_agent_handles_error(agent, agent_input):
"""Test agent handles LLM errors gracefully."""
with patch.object(agent, '_create_llm') as mock_create_llm:
mock_create_llm.side_effect = Exception("LLM unavailable")
result = await agent.execute(agent_input)
assert result["status"] == "error"
assert result["error"] is not None
Run tests:
pytest tests/test_my_agent.py -v