Type-Safe Procurement Agents with Pydantic AI and Merka2a

Pydantic AI brings type-safe, dependency-injected agents to Python. Its @agent.tool decorator validates arguments with Pydantic and passes typed dependencies through a RunContext. Merka2a has no dedicated Pydantic AI package — the REST API is small enough to wrap directly — so a few decorated functions give a fully typed procurement agent.

What We're Building

A pydantic_ai.Agent with an injected HTTP client and three validated tools: search, negotiate, and order.

Prerequisites

• Python 3.10+
• A Merka2a API key (register here)
• An OpenAI API key

Installation

pip install pydantic-ai httpx

Step 1: Typed Dependencies

Pydantic AI passes dependencies into every tool via RunContext. We inject an authenticated httpx client.

import os
from dataclasses import dataclass
import httpx


@dataclass
class Merka2aDeps:
    client: httpx.AsyncClient


def make_client() -> httpx.AsyncClient:
    return httpx.AsyncClient(
        base_url="https://merka2a.com",
        headers={"Authorization": f"Bearer {os.environ['MERKA2A_API_KEY']}"},
        timeout=30.0,
    )

Step 2: Define the Agent

from pydantic_ai import Agent, RunContext

agent = Agent(
    "openai:gpt-4-turbo",
    deps_type=Merka2aDeps,
    system_prompt=(
        "You are a procurement agent for the Merka2a marketplace. "
        "Search, evaluate by price and specs, negotiate when worthwhile, then order. "
        "Prices are in minor units (pence). Each result has an 'offerId'."
    ),
)

Step 3: Register Tools

Each tool receives the typed RunContext[Merka2aDeps] and calls the REST API. Pydantic AI validates the arguments against the function signature.

@agent.tool
async def search_products(
    ctx: RunContext[Merka2aDeps],
    query: str = "",
    category: str = "",
    max_budget: int = 0,
    currency: str = "GBP",
    limit: int = 10,
) -> list:
    """Search the marketplace. max_budget is in minor units (5000 = £50.00)."""
    intent: dict = {"query": query or None, "category": category or None}
    if max_budget:
        intent["budget"] = {"max": {"amount": max_budget, "currency": currency}}
    resp = await ctx.deps.client.post(
        "/v1/search-intent",
        json={"intent": intent, "pagination": {"limit": limit}},
    )
    resp.raise_for_status()
    return resp.json().get("results", [])


@agent.tool
async def negotiate(
    ctx: RunContext[Merka2aDeps],
    offer_id: str,
    target_price: int,
    currency: str = "GBP",
    volume: int = 1,
) -> dict:
    """Negotiate a price (minor units). Returns session id and status."""
    resp = await ctx.deps.client.post(
        "/v1/negotiate",
        json={
            "offerIds": [offer_id],
            "targetPrice": {"amount": target_price, "currency": currency},
            "volume": volume,
        },
    )
    resp.raise_for_status()
    return resp.json()


@agent.tool
async def place_order(
    ctx: RunContext[Merka2aDeps],
    offer_id: str,
    quantity: int = 1,
    negotiation_session_id: str = "",
    shipping_country: str = "GB",
    shipping_city: str = "",
    shipping_postal_code: str = "",
    shipping_method: str = "standard",
) -> dict:
    """Place an order. Pass negotiation_session_id to lock a negotiated price."""
    body: dict = {
        "offerId": offer_id,
        "quantity": quantity,
        "shippingAddress": {
            "country": shipping_country,
            "city": shipping_city or None,
            "postalCode": shipping_postal_code or None,
        },
        "shippingMethod": shipping_method,
    }
    if negotiation_session_id:
        body["negotiationSessionId"] = negotiation_session_id
    resp = await ctx.deps.client.post("/v1/create-order", json=body)
    resp.raise_for_status()
    return resp.json()

Step 4: Run the Agent

import asyncio


async def main():
    async with make_client() as http_client:
        deps = Merka2aDeps(client=http_client)
        result = await agent.run(
            "Find a wireless mouse under £50, negotiate about 10% off, "
            "and ship to London UK (EC1A 1BB).",
            deps=deps,
        )
        print(result.output)


asyncio.run(main())

Why the Typed Approach Helps

Pydantic AI validates every tool argument before the call leaves your process — a model hallucinating a non-numeric max_budget is rejected with a clear error the agent can recover from, rather than producing a malformed API request. The injected httpx client also makes the agent trivial to test: swap in a mock transport in Merka2aDeps.

Next Steps

• LlamaIndex Integration Guide — function tools in LlamaIndex
• CrewAI Integration Guide — role-based crews
• API Reference — full endpoint documentation

---

Ready for a type-safe procurement agent? Get your API key →