Building an Agent From Scratch

A support agent needs to answer questions it can't resolve from a prompt alone: where is this order, and can it be refunded? Give the model functions it can ask to call, run those functions, and send their results back. Keep going until the model has an answer. That's the agent loop.

The interesting decisions are which tools you expose and what their results tell the model. We'll build that loop directly, using small, fake order tools so we can focus on the conversation and the workflow.

Waymark persists the workflow's progress between model and tool calls. If a node crashes, another node resumes from the last saved state. A call that was in flight may run again, so real tools with side effects need to be safe to repeat (idempotent).

Setting up your environment

Start with the Quickstart guide if you haven't run Waymark locally yet. Put this tutorial's workflow and actions in a new Python module.

Stub the workflow

Start with the workflow: it will own the conversation and decide when to call the model or a tool. The actions and data types will follow from what it needs.

from waymark import Workflow, workflow

@workflow
class SupportAgentWorkflow(Workflow):
    async def run(self, message: str) -> str:
        pass

The model needs the conversation so far on every call. Keep that history in run() so it becomes part of the durable workflow state. For now, send the user's message once, append the returned items, and return any text answer.

@workflow
class SupportAgentWorkflow(Workflow):
    async def run(self, message: str) -> str:
        messages: list[ModelItem] = [Message(role="user", content=message)]
        new_items = await roundtrip_agent(messages)
        messages = messages + new_items
        for item in new_items:
            if item.kind == "message":
                return item.content
        return "Tool requests are handled in the next step."

Note the messages = messages + new_items: inside a workflow, build the new list and assign it, rather than handing the list to a helper method that mutates it - a mutation in one function doesn't reach a list held by another (see Known issues).

The model can also return a tool request. The workflow needs to dispatch it to the matching action and attach the call id to the result, so the model knows which request it answers.

@workflow
class SupportAgentWorkflow(Workflow):
    async def run(self, message: str) -> str:
        messages: list[ModelItem] = [Message(role="user", content=message)]
        new_items = await roundtrip_agent(messages)
        messages = messages + new_items
        for item in new_items:
            if item.kind == "message":
                return item.content
        return "Tool requests are handled in the next step."
 
    async def handle_tools(self, request: ToolRequest) -> ToolResult:
        if request.tool_name == "lookup_order":
            content = await lookup_order(request.arguments)
        elif request.tool_name == "refund_order":
            content = await refund_order(request.arguments)
        else:
            content = await unknown_tool(
                UnknownToolInput(tool_name=request.tool_name),
            )
 
        tool_result = ToolResult(
            tool_call_id=request.tool_call_id,
            content=content,
        )
        return tool_result

The tool's arguments go to the action as the plain dict the model produced; the worker validates it against the action's input model before the action runs - see Type hints and validation.

After a tool runs, send its result back to the model. A while True loop keeps the conversation going until the model returns a text answer. This example has no turn limit; add one before using it for open-ended requests.

@workflow
class SupportAgentWorkflow(Workflow):
    async def run(self, message: str) -> str:
        messages: list[ModelItem] = [Message(role="user", content=message)]
 
-        new_items = await roundtrip_agent(messages)
-        messages = messages + new_items
-        for item in new_items:
-            if item.kind == "message":
-                return item.content
-        return "Tool requests are handled in the next step."
 
        while True:
            new_items = await self.run_action(
                roundtrip_agent(messages),
                retry=RetryPolicy(attempts=5, backoff_seconds=10),
            )
            messages = messages + new_items
            for item in new_items:
                if item.kind == "message":
                    return item.content
 
                if item.kind == "tool_request":
                    tool_result = await self.handle_tools(item)
                    messages.append(tool_result)
 
    async def handle_tools(self, request: ToolRequest) -> ToolResult:
        if request.tool_name == "lookup_order":
            content = await lookup_order(request.arguments)
        elif request.tool_name == "refund_order":
            content = await refund_order(request.arguments)
        else:
            content = await unknown_tool(
                UnknownToolInput(tool_name=request.tool_name),
            )
 
        tool_result = ToolResult(
            tool_call_id=request.tool_call_id,
            content=content,
        )
        return tool_result

run() owns the history. The model action returns new items, and the tool helper returns a result for run() to append. Neither needs to mutate a shared list. Now we can define the three kinds of item that pass between them.

Messages and turns

The initial history contains the user's message:

messages: list[ModelItem] = [Message(role="user", content=message)]

A message records who spoke and what they said. The kind field lets the workflow distinguish text from tool requests and results.

from typing import Any, Literal
from pydantic import BaseModel

class Message(BaseModel):
    kind: Literal["message"] = "message"
    role: Literal["user", "assistant"]
    content: str

A tool request names the function, supplies its arguments, and carries an id that we'll copy onto the result.

class ToolRequest(BaseModel):
    kind: Literal["tool_request"] = "tool_request"
    tool_name: str = ""
    tool_call_id: str = ""
    arguments: dict[str, Any]

A tool result pairs that id with the text returned by the action.

class ToolResult(BaseModel):
    kind: Literal["tool_result"] = "tool_result"
    tool_call_id: str
    content: str

ToolResult can be kept small because the workflow doesn't need every internal detail from the order system or billing system. Tool results often work well as JSON strings, but they can be normal text too.

The durable history can contain any of these three item types:

ModelItem = Message | ToolRequest | ToolResult

Tool actions

Each tool is a Waymark action with one Pydantic input model and a string result. The two order tools below return fixed responses; they don't query an order system or issue a refund. That keeps billing credentials and provider setup out of the agent loop we're building.

from waymark import action

class LookupOrderInput(BaseModel):
    order_id: str

class RefundOrderInput(BaseModel):
    order_id: str
    reason: str

class UnknownToolInput(BaseModel):
    tool_name: str

@action
async def lookup_order(params: LookupOrderInput) -> str:
    """Look up the latest status for an order."""
    status = "packed and waiting for carrier pickup"
    return f"Order {params.order_id} is {status}."

@action
async def refund_order(params: RefundOrderInput) -> str:
    """Create a refund for an order."""
    # Replace this with your billing system call.
    return f"Refund for order {params.order_id} was created. Reason: {params.reason}"

@action
async def unknown_tool(params: UnknownToolInput) -> str:
    return f"The tool {params.tool_name!r} is not available."

The input models serve two purposes: the worker validates arguments against them, and we'll use them to tell the model what arguments each tool accepts. Changing the tools is how you adapt this agent to another job.

Round trip the model

The model request belongs in an action because it makes a network call. The loop gives it five attempts, with ten seconds between retries, to recover from failures such as a dropped connection.

The OpenAI client expects message dictionaries. Add a to_remote_message() method to each local item to translate its role, content, or tool call into that format.

import json
 
class Message(BaseModel):
    kind: Literal["message"] = "message"
    role: Literal["user", "assistant"]
    content: str
 
    def to_remote_message(self) -> dict[str, Any]:
        return {"role": self.role, "content": self.content}
 
 
class ToolResult(BaseModel):
    kind: Literal["tool_result"] = "tool_result"
    tool_call_id: str
    content: str
 
    def to_remote_message(self) -> dict[str, Any]:
        return {
            "role": "tool",
            "tool_call_id": self.tool_call_id,
            "content": self.content,
        }
 
 
class ToolRequest(BaseModel):
    kind: Literal["tool_request"] = "tool_request"
    tool_name: str = ""
    tool_call_id: str = ""
    arguments: dict[str, Any]
 
    def to_remote_message(self) -> dict[str, Any]:
        return {
            "role": "assistant",
            "tool_calls": [
                {
                    "id": self.tool_call_id,
                    "type": "function",
                    "function": {
                        "name": self.tool_name,
                        "arguments": json.dumps(self.arguments),
                    },
                }
            ],
        }

The API also needs a JSON Schema for each tool's arguments. Writing a second description by hand would let it drift from the input model. Use Pydantic's model_json_schema() so the schema and validation share one definition.

function_to_schema(...) checks that the function takes exactly one Pydantic model, then uses that model's schema and the function's docstring.

import inspect
from typing import Callable, get_type_hints

def function_to_schema(fn: Callable[..., Any]) -> dict[str, Any]:
    fn = inspect.unwrap(fn)
    parameters = list(inspect.signature(fn).parameters.values())
    if len(parameters) != 1:
        raise TypeError(f"{fn.__name__} must take exactly one input model.")

    parameter = parameters[0]
    input_type = get_type_hints(fn).get(parameter.name)
    if not isinstance(input_type, type) or not issubclass(input_type, BaseModel):
        raise TypeError(f"{fn.__name__} must take one Pydantic model argument.")

    return {
        "type": "function",
        "function": {
            "name": fn.__name__,
            "description": inspect.getdoc(fn) or "",
            "parameters": input_type.model_json_schema(),
        },
    }

call_model(...) sends the history and tool schemas, then translates the reply back into local items. roundtrip_agent(...) exposes that call as a Waymark action and returns only the new items.

from openai import AsyncOpenAI

client = AsyncOpenAI()

SUPPORT_TOOLS = [
    function_to_schema(lookup_order),
    function_to_schema(refund_order),
]

async def call_model(
    messages: list[ModelItem],
    tools: list[dict[str, Any]],
) -> list[ModelItem]:
    response = await client.chat.completions.create(
        model="gpt-5.5",
        messages=[item.to_remote_message() for item in messages],
        tools=tools,
    )

    message = response.choices[0].message
    if message.tool_calls:
        requests: list[ModelItem] = []
        for tool_call in message.tool_calls:
            arguments = json.loads(tool_call.function.arguments or "{}")
            requests.append(
                ToolRequest(
                    tool_name=tool_call.function.name,
                    tool_call_id=tool_call.id,
                    arguments=arguments,
                )
            )
        return requests

    return [Message(role="assistant", content=message.content or "")]

@action
async def roundtrip_agent(messages: list[ModelItem]) -> list[ModelItem]:
    return await call_model(messages, SUPPORT_TOOLS)

For a provider with an OpenAI-compatible API, configure the client with its base_url and api_key.

Run it

Run it like any other workflow:

answer = await SupportAgentWorkflow().run(
    "Where is order A100? If it has not shipped, refund it."
)
print(answer)

The model can now request the order lookup, read its result, and request a refund. Both tools still return fake data; the example makes no changes to an order system.

Where to go next

Keep the loop explicit while you're changing tools and learning what the agent needs. You can see where every message comes from and which action produced each result.

If you already use Pydantic AI, Durable Agents with Pydantic AI explores the same pattern with its Agent abstraction. That tutorial currently has runtime blockers, noted at the top of the page.