Building AI Agents with Function Calling: Tools, Structured Output & Agent Loops

Master function calling for AI agents — from basic tool use to complex multi-step agents with planning, memory, and error recovery.

Introduction

Function calling transforms LLMs from text generators into agents that act. This guide covers production patterns for tool use, structured output, agent loops, and orchestration.

Why Function Calling?

Without Tools	With Function Calling
Text-only responses	Actions in the real world
Hallucinated facts	Verified data from APIs
No memory	Persistent state
Single turn	Multi-step workflows
Manual integration	Standardized interfaces

Part 1: Function Calling Fundamentals

OpenAI Function Calling API

# function_calling/basics.py
from openai import OpenAI
from pydantic import BaseModel, Field
from typing import List, Optional
import json

client = OpenAI()

# Define function schema using Pydantic
class GetWeather(BaseModel):
    """Get current weather for a location."""
    location: str = Field(description="City and state, e.g. 'San Francisco, CA'")
    unit: str = Field(default="celsius", enum=["celsius", "fahrenheit"])

class SearchWeb(BaseModel):
    """Search the web for information."""
    query: str = Field(description="Search query")
    max_results: int = Field(default=5, ge=1, le=10)
    recency_days: Optional[int] = Field(default=None, ge=1, le=365)

# Convert to OpenAI format
def pydantic_to_openai(model: type[BaseModel]) -> dict:
    schema = model.model_json_schema()
    return {
        "type": "function",
        "function": {
            "name": schema["title"],
            "description": schema["description"],
            "parameters": {
                "type": "object",
                "properties": schema["properties"],
                "required": schema.get("required", [])
            }
        }
    }

tools = [pydantic_to_openai(GetWeather), pydantic_to_openai(SearchWeb)]

# Execute function calls
def execute_function(name: str, arguments: dict) -> dict:
    if name == "GetWeather":
        # Call actual weather API
        return {"temperature": 22, "condition": "Sunny", "unit": arguments["unit"]}
    elif name == "SearchWeb":
        # Call search API
        return {"results": [{"title": "Result 1", "url": "..."}]}
    return {"error": f"Unknown function: {name}"}

# Chat completion with tools
def chat_with_tools(messages: list, tools: list) -> dict:
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=messages,
        tools=tools,
        tool_choice="auto",
        parallel_tool_calls=True
    )
    return response.choices[0].message

# Usage
messages = [
    {"role": "user", "content": "What's the weather in Tokyo and search for 'AI agents'"}
]

response_message = chat_with_tools(messages, tools)

if response_message.tool_calls:
    for tool_call in response_message.tool_calls:
        args = json.loads(tool_call.function.arguments)
        result = execute_function(tool_call.function.name, args)

        messages.append(response_message.model_dump())
        messages.append({
            "role": "tool",
            "tool_call_id": tool_call.id,
            "content": json.dumps(result)
        })

    # Get final response
    final = client.chat.completions.create(
        model="gpt-4o",
        messages=messages,
        tools=tools
    )
    print(final.choices[0].message.content)

Structured Output with JSON Schema

# function_calling/structured_output.py
from pydantic import BaseModel, Field, validator
from typing import Literal, List, Optional
from openai import OpenAI
import instructor

client = instructor.from_openai(OpenAI())

class AgentAction(BaseModel):
    """Single action in an agent plan."""
    tool: str = Field(description="Tool name to call")
    args: dict = Field(description="Arguments for the tool")
    reasoning: str = Field(description="Why this action")

class AgentPlan(BaseModel):
    """Complete plan for solving a task."""
    goal: str = Field(description="High-level goal")
    steps: List[AgentAction] = Field(min_items=1)
    estimated_duration_seconds: int = Field(ge=1)
    requires_user_input: bool = False

class AgentResult(BaseModel):
    """Result of agent execution."""
    success: bool
    output: str
    artifacts: List[str] = Field(default_factory=list)
    error: Optional[str] = None
    tokens_used: int = 0

# Use instructor for guaranteed structured output
plan = client.chat.completions.create(
    model="gpt-4o",
    response_model=AgentPlan,
    messages=[
        {"role": "system", "content": "You are a planning agent. Create detailed plans."},
        {"role": "user", "content": "Research the latest LangGraph features and write a summary"}
    ]
)

print(plan.model_dump_json(indent=2))

Part 2: Building a Tool-Enabled Agent

Tool Registry

# agent/tool_registry.py
from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import Any, Callable, Dict, List, Optional
from pydantic import BaseModel
import inspect

@dataclass
class ToolSpec:
    name: str
    description: str
    parameters: dict  # JSON Schema
    function: Callable
    return_type: type

class ToolRegistry:
    def __init__(self):
        self.tools: Dict[str, ToolSpec] = {}

    def register(self, spec: ToolSpec):
        self.tools[spec.name] = spec

    def register_function(
        self,
        name: str,
        description: str,
        func: Callable,
        param_model: type[BaseModel] = None
    ):
        if param_model:
            schema = param_model.model_json_schema()
            parameters = {
                "type": "object",
                "properties": schema.get("properties", {}),
                "required": schema.get("required", [])
            }
        else:
            # Infer from function signature
            sig = inspect.signature(func)
            parameters = {"type": "object", "properties": {}, "required": []}
            for param_name, param in sig.parameters.items():
                if param_name in ("self", "cls"):
                    continue
                parameters["properties"][param_name] = {
                    "type": self._python_type_to_json(param.annotation)
                }
                if param.default == inspect.Parameter.empty:
                    parameters["required"].append(param_name)

        self.register(ToolSpec(
            name=name,
            description=description,
            parameters=parameters,
            function=func,
            return_type=inspect.signature(func).return_annotation
        ))

    def _python_type_to_json(self, annotation) -> str:
        type_map = {
            str: "string", int: "integer", float: "number",
            bool: "boolean", list: "array", dict: "object"
        }
        return type_map.get(annotation, "string")

    def get_openai_tools(self) -> List[dict]:
        return [
            {
                "type": "function",
                "function": {
                    "name": spec.name,
                    "description": spec.description,
                    "parameters": spec.parameters
                }
            }
            for spec in self.tools.values()
        ]

    async def execute(self, name: str, arguments: dict) -> Any:
        if name not in self.tools:
            raise ValueError(f"Unknown tool: {name}")
        spec = self.tools[name]
        return await spec.function(**arguments)

# Example tools
async def web_search(query: str, max_results: int = 5) -> dict:
    # Implementation
    return {"results": []}

async def code_execute(code: str, language: str = "python") -> dict:
    # Implementation with sandbox
    return {"output": "", "error": None}

async def file_read(path: str) -> dict:
    # Implementation
    return {"content": ""}

async def file_write(path: str, content: str) -> dict:
    # Implementation
    return {"success": True}

registry = ToolRegistry()
registry.register_function("web_search", "Search the web", web_search)
registry.register_function("code_execute", "Execute code", code_execute)
registry.register_function("file_read", "Read a file", file_read)
registry.register_function("file_write", "Write a file", file_write)

Agent Core Loop

# agent/core.py
from dataclasses import dataclass, field
from typing import List, Optional, Dict, Any
from openai import OpenAI
import json
import asyncio

@dataclass
class Message:
    role: str
    content: str
    tool_calls: List[dict] = field(default_factory=list)
    tool_call_id: str = ""

@dataclass
class AgentState:
    messages: List[Message]
    tools: ToolRegistry
    max_iterations: int = 10
    current_iteration: int = 0
    context: Dict[str, Any] = field(default_factory=dict)

class ToolAgent:
    def __init__(
        self,
        model: str = "gpt-4o",
        system_prompt: str = "",
        tools: ToolRegistry = None,
        max_iterations: int = 10,
        temperature: float = 0
    ):
        self.client = OpenAI()
        self.model = model
        self.system_prompt = system_prompt
        self.tools = tools or ToolRegistry()
        self.max_iterations = max_iterations
        self.temperature = temperature

    async def run(self, user_input: str, context: dict = None) -> AgentState:
        state = AgentState(
            messages=[
                Message(role="system", content=self.system_prompt),
                Message(role="user", content=user_input)
            ],
            tools=self.tools,
            max_iterations=self.max_iterations,
            context=context or {}
        )

        while state.current_iteration < state.max_iterations:
            state.current_iteration += 1

            # Get model response
            response = await self._get_response(state)

            if not response.tool_calls:
                # No tool calls = done
                state.messages.append(Message(
                    role="assistant",
                    content=response.content
                ))
                break

            # Execute tool calls
            state.messages.append(Message(
                role="assistant",
                content=response.content or "",
                tool_calls=response.tool_calls
            ))

            for tool_call in response.tool_calls:
                result = await self._execute_tool(tool_call)

                state.messages.append(Message(
                    role="tool",
                    content=json.dumps(result),
                    tool_call_id=tool_call["id"]
                ))

        return state

    async def _get_response(self, state: AgentState) -> Any:
        messages = [
            {"role": m.role, "content": m.content, "tool_calls": m.tool_calls}
            for m in state.messages
        ]

        response = self.client.chat.completions.create(
            model=self.model,
            messages=messages,
            tools=self.tools.get_openai_tools(),
            tool_choice="auto",
            temperature=self.temperature,
            parallel_tool_calls=True
        )

        return response.choices[0].message

    async def _execute_tool(self, tool_call: dict) -> dict:
        name = tool_call.function.name
        args = json.loads(tool_call.function.arguments)

        try:
            result = await self.tools.execute(name, args)
            return {"success": True, "result": result}
        except Exception as e:
            return {"success": False, "error": str(e)}

# Usage
SYSTEM_PROMPT = """You are a capable AI agent with access to tools.
Plan your approach, use tools efficiently, and provide clear results.
Always explain your reasoning before taking actions."""

agent = ToolAgent(
    system_prompt=SYSTEM_PROMPT,
    tools=registry,
    max_iterations=15
)

result = await agent.run("Find the latest Python 3.12 features and create a summary file")

Part 3: Advanced Agent Patterns

ReAct Pattern (Reasoning + Acting)

# agent/react.py
from dataclasses import dataclass
from typing import List
import re

REACT_PROMPT = """Answer the following questions as best you can. You have access to the following tools:

{tools}

Use the following format:

Question: the input question you must answer
Thought: you should always think about what to do
Action: the action to take, should be one of [{tool_names}]
Action Input: the input to the action
Observation: the result of the action
... (this Thought/Action/Action Input/Observation can repeat N times)
Thought: I now know the final answer
Final Answer: the final answer to the original input question

Begin!

Question: {input}
Thought:"""

class ReActAgent:
    def __init__(self, llm, tools: ToolRegistry, max_steps: int = 10):
        self.llm = llm
        self.tools = tools
        self.max_steps = max_steps

    async def run(self, question: str) -> str:
        tool_descriptions = "\n".join(
            f"{spec.name}: {spec.description}" 
            for spec in self.tools.tools.values()
        )
        tool_names = ", ".join(self.tools.tools.keys())

        prompt = REACT_PROMPT.format(
            tools=tool_descriptions,
            tool_names=tool_names,
            input=question
        )

        for step in range(self.max_steps):
            response = await self.llm.ainvoke(prompt)

            # Parse action
            action_match = re.search(r"Action:\s*(\w+)", response)
            input_match = re.search(r"Action Input:\s*(.+)", response)
            final_match = re.search(r"Final Answer:\s*(.+)", response, re.DOTALL)

            if final_match:
                return final_match.group(1).strip()

            if action_match and input_match:
                tool_name = action_match.group(1)
                tool_input = json.loads(input_match.group(1))

                result = await self.tools.execute(tool_name, tool_input)

                prompt += f"\n{response}\nObservation: {json.dumps(result)}\nThought:"
            else:
                prompt += f"\n{response}\nThought:"

        return "Max steps reached without final answer"

Plan-and-Execute Agent

# agent/plan_execute.py
from dataclasses import dataclass
from typing import List, Optional
from pydantic import BaseModel, Field

class PlanStep(BaseModel):
    step_number: int
    description: str
    tool: str
    args: dict
    expected_output: str
    depends_on: List[int] = Field(default_factory=list)

class ExecutionPlan(BaseModel):
    goal: str
    steps: List[PlanStep]
    success_criteria: str

class PlanAndExecuteAgent:
    def __init__(self, planner_llm, executor_llm, tools: ToolRegistry):
        self.planner = planner_llm
        self.executor = executor_llm
        self.tools = tools

    async def run(self, goal: str) -> dict:
        # Phase 1: Plan
        plan = await self._create_plan(goal)

        # Phase 2: Execute
        results = {}
        for step in plan.steps:
            # Check dependencies
            for dep in step.depends_on:
                if dep not in results:
                    raise ValueError(f"Dependency step {dep} not completed")

            # Execute step
            result = await self._execute_step(step, results)
            results[step.step_number] = result

            # Check if we should continue
            if not result.get("success"):
                return {"success": False, "plan": plan, "results": results, "failed_step": step.step_number}

        # Phase 3: Verify
        verification = await self._verify(plan, results)

        return {
            "success": verification["success"],
            "plan": plan,
            "results": results,
            "verification": verification
        }

    async def _create_plan(self, goal: str) -> ExecutionPlan:
        planning_prompt = f"""Create a detailed execution plan for: {goal}

Available tools: {list(self.tools.tools.keys())}

Return a plan with numbered steps, each specifying:
- tool to use
- arguments
- expected output
- dependencies on other steps

Output as JSON matching ExecutionPlan schema."""

        return await self.planner.with_structured_output(ExecutionPlan).ainvoke(planning_prompt)

    async def _execute_step(self, step: PlanStep, previous_results: dict) -> dict:
        # Substitute variables from previous results
        args = self._substitute_variables(step.args, previous_results)

        try:
            result = await self.tools.execute(step.tool, args)
            return {"success": True, "output": result}
        except Exception as e:
            return {"success": False, "error": str(e)}

    def _substitute_variables(self, args: dict, results: dict) -> dict:
        # Replace {{step_N.output.field}} with actual values
        import re
        def replace(match):
            ref = match.group(1)
            step_num, *path = ref.split(".")
            step_num = int(step_num)
            if step_num in results:
                value = results[step_num]["output"]
                for p in path:
                    value = value.get(p, "")
                return str(value)
            return match.group(0)

        substituted = {}
        for k, v in args.items():
            if isinstance(v, str):
                substituted[k] = re.sub(r"\{\{step_(\d+(?:\.\w+)*)\}\}", replace, v)
            else:
                substituted[k] = v
        return substituted

Self-Correcting Agent

# agent/self_correcting.py
from dataclasses import dataclass
from typing import Optional
import asyncio

@dataclass
class Attempt:
    iteration: int
    plan: str
    actions: list
    result: dict
    error: Optional[str] = None
    reflection: str = ""

class SelfCorrectingAgent:
    def __init__(
        self,
        llm,
        tools: ToolRegistry,
        max_attempts: int = 3,
        reflection_llm = None
    ):
        self.llm = llm
        self.tools = tools
        self.max_attempts = max_attempts
        self.reflection_llm = reflection_llm or llm

    async def run(self, goal: str) -> dict:
        attempts = []

        for attempt_num in range(1, self.max_attempts + 1):
            # Create plan (incorporating reflections from previous attempts)
            plan = await self._create_plan(goal, attempts)

            # Execute
            result = await self._execute_plan(plan)

            attempt = Attempt(
                iteration=attempt_num,
                plan=plan,
                actions=result.get("actions", []),
                result=result,
                error=result.get("error")
            )
            attempts.append(attempt)

            if result.get("success"):
                return {"success": True, "attempts": attempts, "final_result": result}

            # Reflect on failure
            if attempt_num < self.max_attempts:
                reflection = await self._reflect(goal, attempt)
                attempt.reflection = reflection

        return {"success": False, "attempts": attempts}

    async def _create_plan(self, goal: str, attempts: list) -> str:
        reflection_context = ""
        if attempts:
            reflection_context = "\n\nPrevious attempts and reflections:\n"
            for a in attempts:
                reflection_context += f"Attempt {a.iteration}: {a.reflection}\n"

        prompt = f"""Goal: {goal}
{reflection_context}

Create a plan to achieve the goal. Learn from previous failures.
Available tools: {list(self.tools.tools.keys())}"""

        return await self.llm.ainvoke(prompt)

    async def _execute_plan(self, plan: str) -> dict:
        # Execute using tool agent
        agent = ToolAgent(llm=self.llm, tools=self.tools)
        state = await agent.run(plan)

        # Check success
        last_msg = state.messages[-1]
        return {
            "success": "error" not in last_msg.content.lower() if last_msg.content else False,
            "actions": [m.tool_calls for m in state.messages if m.tool_calls],
            "final_output": last_msg.content
        }

    async def _reflect(self, goal: str, attempt: Attempt) -> str:
        prompt = f"""Goal: {goal}
Plan: {attempt.plan}
Result: {attempt.result}
Error: {attempt.error}

Analyze why this failed and provide specific guidance for the next attempt.
Focus on: wrong tool choice, incorrect arguments, missing steps, logic errors."""

        return await self.reflection_llm.ainvoke(prompt)

Part 4: Multi-Tool Workflows

Parallel Tool Execution

# agent/parallel.py
import asyncio
from dataclasses import dataclass
from typing import List

@dataclass
class ParallelAction:
    tool: str
    args: dict
    dependency: str = None  # Tool name this depends on

class ParallelAgent:
    def __init__(self, llm, tools: ToolRegistry):
        self.llm = llm
        self.tools = tools

    async def execute_parallel(self, actions: List[ParallelAction]) -> dict:
        """Execute independent actions in parallel."""
        # Group by dependencies
        independent = [a for a in actions if not a.dependency]
        dependent = [a for a in actions if a.dependency]

        # Execute independent first
        results = {}
        if independent:
            tasks = [self.tools.execute(a.tool, a.args) for a in independent]
            outputs = await asyncio.gather(*tasks, return_exceptions=True)
            for action, output in zip(independent, outputs):
                results[action.tool] = output if not isinstance(output, Exception) else {"error": str(output)}

        # Execute dependent
        for action in dependent:
            dep_result = results.get(action.dependency)
            if dep_result and "error" not in dep_result:
                # Substitute dependency result
                args = self._inject_dependency(action.args, action.dependency, dep_result)
                output = await self.tools.execute(action.tool, args)
                results[action.tool] = output
            else:
                results[action.tool] = {"error": f"Dependency {action.dependency} failed"}

        return results

Tool Chaining with Validation

# agent/chaining.py
from pydantic import BaseModel, field_validator
from typing import Any, Callable

class ToolChainStep(BaseModel):
    tool: str
    args: dict
    validator: Optional[Callable[[Any], bool]] = None
    on_failure: str = "stop"  # "stop", "retry", "fallback"
    fallback: Optional["ToolChainStep"] = None

class ToolChain:
    def __init__(self, tools: ToolRegistry):
        self.tools = tools
        self.steps: List[ToolChainStep] = []

    def add_step(self, step: ToolChainStep):
        self.steps.append(step)
        return self

    async def execute(self, initial_input: dict) -> dict:
        context = {"input": initial_input}

        for i, step in enumerate(self.steps):
            # Prepare args with context
            args = self._prepare_args(step.args, context)

            # Execute with retries
            max_retries = 2 if step.on_failure == "retry" else 0

            for attempt in range(max_retries + 1):
                try:
                    result = await self.tools.execute(step.tool, args)

                    # Validate
                    if step.validator and not step.validator(result):
                        raise ValueError(f"Validation failed for step {i}")

                    context[f"step_{i}"] = result
                    break

                except Exception as e:
                    if attempt == max_retries:
                        if step.on_failure == "fallback" and step.fallback:
                            result = await self._execute_fallback(step.fallback, context)
                            context[f"step_{i}"] = result
                            break
                        else:
                            return {"success": False, "error": str(e), "step": i, "context": context}

        return {"success": True, "context": context}

Part 5: Memory & State Management

Conversation Memory

# agent/memory.py
from dataclasses import dataclass, field
from typing import List, Dict, Any
from datetime import datetime
import json

@dataclass
class MemoryEntry:
    timestamp: str
    type: str  # "user", "assistant", "tool", "observation"
    content: str
    metadata: Dict[str, Any] = field(default_factory=dict)

class ConversationMemory:
    def __init__(self, max_entries: int = 100):
        self.entries: List[MemoryEntry] = []
        self.max_entries = max_entries
        self.summary: str = ""

    def add(self, entry: MemoryEntry):
        self.entries.append(entry)
        if len(self.entries) > self.max_entries:
            self._summarize_old()

    def _summarize_old(self):
        # Keep recent entries, summarize older ones
        keep = self.entries[-20:]
        old = self.entries[:-20]

        if old:
            # Generate summary of old entries
            old_text = "\n".join(f"{e.type}: {e.content}" for e in old)
            self.summary = f"Previous conversation summary: {old_text[:2000]}..."

        self.entries = keep

    def get_context(self, max_tokens: int = 4000) -> str:
        parts = []
        if self.summary:
            parts.append(self.summary)

        for entry in self.entries:
            parts.append(f"{entry.type}: {entry.content}")

        context = "\n".join(parts)
        # Rough token estimation
        if len(context) > max_tokens * 4:
            # Truncate older entries
            return context[-max_tokens * 4:]
        return context

    def get_relevant(self, query: str, k: int = 5) -> List[MemoryEntry]:
        # Simple keyword matching (replace with embeddings for production)
        query_words = set(query.lower().split())
        scored = []
        for entry in self.entries:
            content_words = set(entry.content.lower().split())
            overlap = len(query_words & content_words)
            if overlap > 0:
                scored.append((overlap, entry))

        scored.sort(reverse=True)
        return [e for _, e in scored[:k]]

Persistent State (SQLite)

# agent/state.py
import sqlite3
import json
from dataclasses import dataclass
from typing import Any, Optional
from datetime import datetime

@dataclass
class AgentState:
    session_id: str
    agent_type: str
    state_data: dict
    created_at: str
    updated_at: str

class StateManager:
    def __init__(self, db_path: str = "agent_state.db"):
        self.conn = sqlite3.connect(db_path)
        self._init_db()

    def _init_db(self):
        self.conn.execute("""
            CREATE TABLE IF NOT EXISTS agent_states (
                session_id TEXT PRIMARY KEY,
                agent_type TEXT,
                state_data TEXT,
                created_at TEXT,
                updated_at TEXT
            )
        """)
        self.conn.execute("""
            CREATE TABLE IF NOT EXISTS agent_logs (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                session_id TEXT,
                timestamp TEXT,
                event_type TEXT,
                data TEXT,
                FOREIGN KEY (session_id) REFERENCES agent_states(session_id)
            )
        """)
        self.conn.commit()

    def save_state(self, state: AgentState):
        self.conn.execute("""
            INSERT OR REPLACE INTO agent_states 
            (session_id, agent_type, state_data, created_at, updated_at)
            VALUES (?, ?, ?, ?, ?)
        """, (
            state.session_id,
            state.agent_type,
            json.dumps(state.state_data),
            state.created_at,
            state.updated_at
        ))
        self.conn.commit()

    def load_state(self, session_id: str) -> Optional[AgentState]:
        cursor = self.conn.execute(
            "SELECT * FROM agent_states WHERE session_id = ?", (session_id,)
        )
        row = cursor.fetchone()
        if row:
            return AgentState(
                session_id=row[0],
                agent_type=row[1],
                state_data=json.loads(row[2]),
                created_at=row[3],
                updated_at=row[4]
            )
        return None

    def log_event(self, session_id: str, event_type: str, data: dict):
        self.conn.execute(
            "INSERT INTO agent_logs (session_id, timestamp, event_type, data) VALUES (?, ?, ?, ?)",
            (session_id, datetime.utcnow().isoformat(), event_type, json.dumps(data))
        )
        self.conn.commit()

    def get_logs(self, session_id: str) -> list:
        cursor = self.conn.execute(
            "SELECT * FROM agent_logs WHERE session_id = ? ORDER BY timestamp",
            (session_id,)
        )
        return cursor.fetchall()

Part 6: Production Deployment

FastAPI Agent Server

# server/agent_server.py
from fastapi import FastAPI, HTTPException, BackgroundTasks
from pydantic import BaseModel
from typing import Optional, Dict, Any
import uuid

app = FastAPI(title="AI Agent API")

class RunRequest(BaseModel):
    goal: str
    agent_type: str = "tool_agent"
    context: Optional[Dict[str, Any]] = None
    max_iterations: int = 10
    session_id: Optional[str] = None

class RunResponse(BaseModel):
    session_id: str
    success: bool
    result: Optional[str] = None
    error: Optional[str] = None
    steps: int
    tokens_used: int

@app.post("/agent/run", response_model=RunResponse)
async def run_agent(request: RunRequest, background_tasks: BackgroundTasks):
    session_id = request.session_id or str(uuid.uuid4())

    # Load or create agent
    agent = get_agent(request.agent_type)

    # Load previous state if exists
    state = state_manager.load_state(session_id) if request.session_id else None

    try:
        result = await agent.run(request.goal, context=request.context)

        # Save state
        background_tasks.add_task(save_state, session_id, result)

        return RunResponse(
            session_id=session_id,
            success=True,
            result=result.messages[-1].content if result.messages else None,
            steps=result.current_iteration,
            tokens_used=calculate_tokens(result.messages)
        )
    except Exception as e:
        return RunResponse(
            session_id=session_id,
            success=False,
            error=str(e),
            steps=0,
            tokens_used=0
        )

@app.get("/agent/state/{session_id}")
async def get_state(session_id: str):
    state = state_manager.load_state(session_id)
    if not state:
        raise HTTPException(404, "Session not found")
    return state.state_data

@app.get("/agent/logs/{session_id}")
async def get_logs(session_id: str):
    return state_manager.get_logs(session_id)

Docker Deployment

# Dockerfile
FROM python:3.11-slim

WORKDIR /app

# Install dependencies
RUN apt-get update && apt-get install -y \
    curl \
    && rm -rf /var/lib/apt/lists/*

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

# Non-root user
RUN useradd -m -u 1000 agent && chown -R agent:agent /app
USER agent

HEALTHCHECK --interval=30s --timeout=10s --start-period=5s --retries=3 \
    CMD curl -f http://localhost:8000/health || exit 1

CMD ["uvicorn", "server.agent_server:app", "--host", "0.0.0.0", "--port", "8000"]

Part 7: Testing & Evaluation

Agent Evaluation Framework

# evaluation/agent_eval.py
from dataclasses import dataclass
from typing import List, Callable
import asyncio

@dataclass
class AgentTestCase:
    name: str
    goal: str
    expected_outcome: str
    expected_tools: List[str] = None
    max_steps: int = 10
    setup: Callable = None
    teardown: Callable = None

class AgentEvaluator:
    def __init__(self, agent_factory: Callable):
        self.agent_factory = agent_factory

    async def evaluate(self, test_cases: List[AgentTestCase]) -> dict:
        results = []

        for tc in test_cases:
            if tc.setup:
                await tc.setup()

            agent = self.agent_factory()
            result = await agent.run(tc.goal)

            # Evaluate
            eval_result = await self._evaluate_single(tc, result)
            results.append(eval_result)

            if tc.teardown:
                await tc.teardown()

        return self._aggregate(results)

    async def _evaluate_single(self, tc: AgentTestCase, result) -> dict:
        # Check success
        success = result.get("success", False)

        # Check tool usage
        tools_used = set()
        for msg in result.get("messages", []):
            if msg.get("tool_calls"):
                for tc_call in msg["tool_calls"]:
                    tools_used.add(tc_call["function"]["name"])

        expected_tools = set(tc.expected_tools or [])
        tool_precision = len(tools_used & expected_tools) / max(1, len(tools_used))
        tool_recall = len(tools_used & expected_tools) / max(1, len(expected_tools))

        # LLM judge for outcome quality
        quality = await self._llm_judge(tc.goal, tc.expected_outcome, result)

        return {
            "name": tc.name,
            "success": success,
            "steps": result.get("steps", 0),
            "tool_precision": tool_precision,
            "tool_recall": tool_recall,
            "quality_score": quality
        }

    async def _llm_judge(self, goal: str, expected: str, actual) -> float:
        # Use LLM to evaluate quality
        prompt = f"""Rate the agent's result on a scale of 0-1.

Goal: {goal}
Expected: {expected}
Actual: {actual}

Consider: completeness, accuracy, efficiency, clarity.
Return only a number between 0 and 1."""

        response = await self.llm.ainvoke(prompt)
        try:
            return float(response.strip())
        except:
            return 0.5

Decision Framework

FUNCTION CALLING PATTERN SELECTION:

Simple Q&A with tools?
├─→ Yes → Basic function calling (single turn)
└─→ No
    ├─→ Multi-step with known sequence?
    │   ├─→ Yes → Plan-and-Execute
    │   └─→ No
    │       ├─→ Need to adapt based on results?
    │       │   ├─→ Yes → ReAct or Self-Correcting
    │       │   └─→ No → Tool Chain
    │       │
    │       ├─→ Multiple independent tools?
    │       │   └─→ Yes → Parallel execution
    │       │
    │       └─→ Complex, evolving task → Hybrid (Plan + ReAct + Self-Correct)

Production Checklist

Function Definitions

• [ ] All tools have clear descriptions and examples
• [ ] Parameter schemas are strict (Pydantic validated)
• [ ] Tool responses are structured, not free text
• [ ] Error responses include actionable information
• [ ] Idempotent tools marked for safe retry

Agent Reliability

• [ ] Max iteration limits prevent infinite loops
• [ ] Timeout handling for each tool call
• [ ] Circuit breakers for external APIs
• [ ] Graceful degradation when tools fail
• [ ] State persistence for long-running agents

Observability

• [ ] Tool call logging (input, output, latency, errors)
• [ ] Token usage tracking per step
• [ ] Cost tracking per agent run
• [ ] Success/failure rates by tool
• [ ] Latency percentiles (p50, p95, p99)

Security

• [ ] Input validation on all tool parameters
• [ ] Sandbox for code execution tools
• [ ] Rate limiting on external API calls
• [ ] Authentication for sensitive tools
• [ ] Audit logging for all actions

Testing

• [ ] Unit tests for each tool
• [ ] Integration tests for agent flows
• [ ] Golden test cases for common tasks
• [ ] Chaos testing (tool failures, timeouts)
• [ ] Regression tests in CI/CD

Resources

• OpenAI Function Calling
• Instructor Library
• LangGraph Agents
• ReAct Paper
• Agent Patterns
• Function Calling Best Practices

---

Function calling is the bridge between LLM reasoning and real-world action. Build robust tool interfaces, implement proper error handling, and always evaluate your agents on realistic tasks before deploying.