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