Home › Fp For Ai Agents › Article 3
Part 3 AI Agents

Monads as Agent Protocol

Image: AI-generated

Monads as Agent Protocol

Why Result, Maybe, and Validation are the universal control protocol for AI agent workflows.

In the previous article we learned about functors: They transform values within a context. But what if the transformation itself produces a new context?

def parse_int(s: str) -> Maybe:
    ...  # Returns Just(42) or Nothing()

Just("42").map(parse_int)  # → Just(Just(42)) – nested!

The problem: map wraps the result again. We need an operation that flattens the context: bind (also called flatMap or >>=). And that's exactly what a monad does.

1. Q: What Is a Monad?

A: A monoid in the category of endofunctors.

This statement is worth keeping in mind for later, but for now let's continue with the practical definition:

A monad is a functor with an additional operation called bind or flatMap:

Plus three laws:

  1. Left identity: return(a).bind(f) == f(a)
  2. Right identity: m.bind(return) == m
  3. Associativity: m.bind(f).bind(g) == m.bind(λx. f(x).bind(g))

This sounds very abstract, but the intuition is simple: A monad is a value in context with the ability to chain operations while the context is managed automatically.

"Monads let us compose functions that have effects. The key insight is that the plumbing is hidden — you only see the happy path." — Bartosz Milewski: Category Theory for Programmers, p. 317

"A monad is just a monoid in the category of endofunctors." — Saunders Mac Lane: Categories for the Working Mathematician

2. Result as Agent Protocol

The Result monad is the central protocol for agent workflows:

@dataclass(frozen=True)
class Ok:
    value: any
    def map(self, f):  return Ok(f(self.value))
    def bind(self, f): return f(self.value)

@dataclass(frozen=True)
class Err:
    error: str
    def map(self, f):  return self  # Pass error through
    def bind(self, f): return self  # Pass error through

Railway Oriented Programming

The beauty: On Err, every subsequent step is skipped. This is Railway Oriented Programming — the error flows through automatically:

def validate_input(data: dict) -> Result:
    if "action" not in data:
        return Err("Field 'action' is missing")
    return Ok(data)

def normalize_data(data: dict) -> Result:
    return Ok({k.lower().strip(): v for k, v in data.items()})

def enrich_with_metadata(data: dict) -> Result:
    return Ok({**data, "processed": True})

# Pipeline: validate → normalize → enrich
pipeline = pipeline(validate_input, normalize_data, enrich_with_metadata)

pipeline({"action": "search"})  # → Ok({...})
pipeline({"no_action": True})   # → Err("Field 'action' is missing")

For the AI agent, this is ideal: It doesn't need to handle errors itself. The first error stops the pipeline, and it gets a clear error message.

Verifying Monad Laws

# Left identity: Ok(5).bind(f) == f(5)
f = lambda x: Ok(x * 2)
assert Ok(5).bind(f) == f(5)  # Ok(10) == Ok(10) ✓

# Right identity: m.bind(Ok) == m
assert Ok(5).bind(Ok) == Ok(5)  # ✓
assert Err("e").bind(Ok) == Err("e")  # ✓

# Associativity: m.bind(f).bind(g) == m.bind(λx. f(x).bind(g))
g = lambda x: Ok(x + 1)
assert Ok(5).bind(f).bind(g) == Ok(5).bind(lambda x: f(x).bind(g))  # ✓

Agent Workflow Orchestration

def run_workflow(steps, initial):
    """Executes named steps — stops at first error."""
    result = Ok(initial)
    log = []
    for name, fn in steps:
        result = result.bind(fn)
        if result.is_ok:
            log.append(f"✓ {name}")
        else:
            log.append(f"✗ {name}: {result.error}")
            break
    return WorkflowResult(steps_completed=..., final_result=result, log=tuple(log))

→ Full code: src/agent_result.py

3. Maybe for Optional Data

The Maybe monad models optional values — without None/Null chaos:

@dataclass(frozen=True)
class Just:
    value: any
    def bind(self, f): return f(self.value)
    def or_else(self, default): return self.value

@dataclass(frozen=True)
class Nothing:
    def bind(self, f): return self
    def or_else(self, default): return default

Agent Use Cases

Safe dictionary access:

def from_dict(d: dict, key: str) -> Maybe:
    if key in d:
        return Just(d[key])
    return Nothing()

# No KeyError, no try/except
from_dict(config, "model").or_else("gpt-4")

Safe parsing:

def safe_parse_int(s: str) -> Maybe:
    try:
        return Just(int(s))
    except (ValueError, TypeError):
        return Nothing()

safe_parse_int("42")   # → Just(42)
safe_parse_int("abc")  # → Nothing()

Fallback chains:

def chain_lookups(*lookups):
    """Returns the first Just."""
    for lookup in lookups:
        result = lookup()
        if isinstance(result, Just):
            return result
    return Nothing()

# Agent searches: first cache, then DB, then default
chain_lookups(
    lambda: from_dict(cache, "key"),
    lambda: from_dict(database, "key"),
    lambda: Just("default"),
)

Immutable configuration:

@dataclass(frozen=True)
class AgentConfig:
    settings: tuple[tuple[str, str], ...] = ()

    def get(self, key: str) -> Maybe:
        for k, v in self.settings:
            if k == key: return Just(v)
        return Nothing()

    def with_setting(self, key: str, value: str) -> AgentConfig:
        # Returns new config — original remains unchanged
        ...

→ Full code: src/agent_maybe.py

4. Validation for Parallel Error Collection

Result stops at the first error. But sometimes you want all errors at once:

@dataclass(frozen=True)
class Valid:
    value: any

@dataclass(frozen=True)
class Invalid:
    errors: tuple[str, ...]

Applicative Combination

def combine(*validations):
    """All Valid → Valid(tuple), at least one Invalid → Invalid(all errors)."""
    errors, values = [], []
    for v in validations:
        match v:
            case Valid(value): values.append(value)
            case Invalid(errs): errors.extend(errs)
    if errors:
        return Invalid(tuple(errors))
    return Valid(tuple(values))

Agent Use Case: Validating Tool Input

def validate_tool_input(data: dict) -> Validation:
    action_v = validate_not_empty("action", data.get("action", ""))
    target_v = validate_not_empty("target", data.get("target", ""))
    return combine_with(
        lambda a, t: ToolInput(action=a, target=t),
        action_v, target_v,
    )

# All errors at once:
validate_tool_input({})
# → Invalid(("action must not be empty", "target must not be empty"))

For the AI agent, this is especially valuable: Instead of correcting one error at a time, it gets all problems at once and can fix them in a single step.

→ Full code: src/agent_validation.py

5. Monads as a Universal Agent Interface

All three monads follow the same interface:

Monad map bind Context
Result Transforms Ok value Chains steps, Err stops Success/Error
Maybe Transforms Just value Chains lookups, Nothing stops Present/Not present
Validation Transforms Valid value — (Applicative instead of Monad) Valid/All errors

The AI agent only needs to know one pattern:

step1(input).bind(step2).bind(step3)

And make three decisions:

  1. Result → when errors should stop the workflow
  2. Maybe → when missing values are ok (with fallback)
  3. Validation → when all errors should be collected

The Central Insight

Monad = protocol for "what happens on success/failure".

The semantics live in the data structure, not in control flow. The agent needs no if/else chains, no try/except blocks — the monad handles it automatically.

Try It Yourself

The Railway-Oriented Programming pattern from section 2 can be tried in a few lines: Ok/Err with map/bind, three tiny pipeline steps, and a generic pipeline function that chains them. Note: the pipeline code contains not a single if/else error branch – Err passes itself through.

from dataclasses import dataclass

@dataclass(frozen=True)
class Ok:
    value: object
    def map(self, f):  return Ok(f(self.value))
    def bind(self, f): return f(self.value)

@dataclass(frozen=True)
class Err:
    error: str
    def map(self, f):  return self
    def bind(self, f): return self

def validate_input(data):
    if "action" not in data:
        return Err("field 'action' missing")
    return Ok(data)

def normalize_data(data):
    return Ok({k.lower().strip(): v for k, v in data.items()})

def enrich_with_metadata(data):
    return Ok({**data, "processed": True})

def pipeline(*steps):
    """Chains steps of type A -> Result[B]."""
    def run(initial):
        result = Ok(initial)
        for step in steps:
            result = result.bind(step)
        return result
    return run

run = pipeline(validate_input, normalize_data, enrich_with_metadata)

for case in [
    {"action": "Search", "Query": "functor"},
    {"no_action": True},
    {"action": "SUMMARIZE", "TARGET": "doc.pdf"},
]:
    print(f"{case!r:55} → {run(case)}")

Extension idea: add a step log_result(name) that uses map to append an entry to a processed_log – map stays pure because the whole state is passed on immutably, and the pipeline itself doesn't change.

Outlook

Now we have the building blocks: Pure functions, immutability, functors, monads. In the next article we put it all together: composition, configurable pipelines, tool registry, and automatic code validation.

Sources

Code Examples

📄 As PDF