Skip to main content

Error Codes Reference

All AgentFlow exceptions use structured error codes for programmatic handling and observability.

Exception Hierarchy


Graph Errors

GRAPH_000 — Generic Graph Error

Base error code for general graph operations.

FieldValue
CodeGRAPH_000
RetryableNo
CategoryGraph

Raised by: Any graph operation failure.

Example:

from agentflow.core.exceptions import GraphError

raise GraphError(
message="Graph failed to initialize",
error_code="GRAPH_000",
context={"graph_name": "my_agent"}
)

Common causes:

  • Invalid graph configuration
  • Node initialization failure
  • Edge routing error

GRAPH_001+ — Extended Graph Errors

Extended error codes for specific graph failures.

FieldValue
CodesGRAPH_001, GRAPH_002, etc.
RetryableNo
CategoryGraph

Node Errors

NODE_000 — Generic Node Error

Base error code for node-specific errors.

FieldValue
CodeNODE_000
RetryableDepends on cause
CategoryGraph / Node

Raised by: Node execution failures.

Example:

from agentflow.core.exceptions import NodeError

raise NodeError(
message="Node execution failed",
error_code="NODE_000",
context={"node_name": "process_data", "input_size": 100}
)

Common causes:

  • Tool execution failure
  • Invalid node input
  • Node timeout
  • Node resource exhaustion

NODE_001+ — Extended Node Errors

Extended error codes for specific node failures.

FieldValue
CodesNODE_001, NODE_002, etc.
RetryableDepends on cause

Recursion Errors

RECURSION_000 — Recursion Limit Exceeded

Raised when graph execution exceeds the configured recursion limit.

FieldValue
CodeRECURSION_000
RetryableNo (requires config change)
CategoryGraph

Example:

from agentflow.core.exceptions import GraphRecursionError

raise GraphRecursionError(
message="Recursion limit exceeded in graph execution",
error_code="RECURSION_000",
context={"recursion_depth": 100, "max_depth": 50}
)

Common causes:

  • Infinite loop in graph routing
  • Tool repeatedly calling itself
  • Missing termination condition
  • Tool selection cycling

Fix: Check your graph for loops, add recursion limits, or increase recursion_limit in config.


Storage Errors

STORAGE_000 — Generic Storage Error

Base error code for storage layer errors.

FieldValue
CodeSTORAGE_000
RetryableNo
CategoryStorage

Raised by: General storage operation failures.


STORAGE_TRANSIENT_000 — Transient Storage Error

Temporary storage failures that may succeed on retry.

FieldValue
CodeSTORAGE_TRANSIENT_000
RetryableYes
CategoryStorage

Common causes:

  • Database connection timeout
  • Network interruption
  • Lock contention
  • Temporary resource unavailability

Example:

from agentflow.core.exceptions import TransientStorageError

raise TransientStorageError(
message="Database connection timeout",
error_code="STORAGE_TRANSIENT_000",
context={"operation": "read_thread", "timeout_ms": 5000}
)

Recovery: Implement exponential backoff retry logic.


STORAGE_SERIALIZATION_000 — Serialization Error

Failed to serialize or deserialize data.

FieldValue
CodeSTORAGE_SERIALIZATION_000
RetryableNo (data is corrupt/invalid)
CategoryStorage

Common causes:

  • Invalid state schema
  • Corrupt checkpoint data
  • Incompatible schema version
  • Invalid message format

Example:

from agentflow.core.exceptions import SerializationError

raise SerializationError(
message="Failed to deserialize state",
error_code="STORAGE_SERIALIZATION_000",
context={"thread_id": "abc123", "schema_version": "2"}
)

STORAGE_SCHEMA_000 — Schema Version Error

Schema version detection or migration failed.

FieldValue
CodeSTORAGE_SCHEMA_000
RetryableNo (requires migration)
CategoryStorage

Common causes:

  • Upgrading AgentFlow without running migrations
  • Database schema out of sync with code version
  • Corrupt schema version table

Fix: Run database migrations after upgrading AgentFlow.


STORAGE_NOT_FOUND_000 — Resource Not Found

Requested resource does not exist in storage.

FieldValue
CodeSTORAGE_NOT_FOUND_000
RetryableNo (resource doesn't exist)
CategoryStorage

Common causes:

  • Invalid thread_id
  • Thread was deleted
  • Checkpoint expired
  • Missing required resource

Example:

from agentflow.core.exceptions import ResourceNotFoundError

raise ResourceNotFoundError(
message="Thread not found",
error_code="STORAGE_NOT_FOUND_000",
context={"thread_id": "abc123"}
)

METRICS_000 — Metrics Error

Non-critical metrics emission failure.

FieldValue
CodeMETRICS_000
RetryableNo (metrics are non-critical)
CategoryObservability

Note: This error is typically logged but not raised, as metrics failures should not interrupt operations.


Media Errors

MEDIA_000 — Unsupported Media Input

Model cannot accept the given media input type.

FieldValue
CodeMEDIA_000 (via exception)
RetryableNo (requires model change)
CategoryMedia

Raised by: UnsupportedMediaInputError

Attributes:

AttributeTypeDescription
providerstringProvider identifier (e.g., "openai", "google")
modelstringModel name (e.g., "gpt-4o", "gemini-1.5-pro")
media_typestringType of media (e.g., "image", "document", "audio")
source_kindstringHow media was provided ("url", "file_id", "data", "internal_ref")
transports_attemptedlistTransport modes tried before failure

Common causes:

  • Model doesn't support vision but image was provided
  • Model doesn't support document input
  • External URL not allowed for provider

Example:

from agentflow.core.exceptions import UnsupportedMediaInputError

raise UnsupportedMediaInputError(
provider="openai",
model="gpt-4o-mini",
media_type="image",
source_kind="url",
transports_attempted=["inline", "url"]
)

Fix: Use a vision-capable model (e.g., gpt-4o, gemini-1.5-pro) or remove media inputs.


Validation Errors

VALIDATION_000 — Input Validation Failed

Input validation detected a policy violation.

FieldValue
CodeVALIDATION_000 (via exception)
RetryableNo (input is invalid)
CategorySecurity

Raised by: ValidationError

Attributes:

AttributeTypeDescription
messagestringHuman-readable error message
violation_typestringType of violation
detailsdictAdditional violation details

Common violation types:

Violation TypeDescription
prompt_injectionDirect or indirect prompt injection detected
jailbreakJailbreak attempt detected
content_policyContent policy violation
encoding_attackObfuscation via encoding detected
delimiter_confusionConflicting delimiters in input
payload_splittingAttack distributed across inputs
system_leakAttempt to extract system prompt

Example:

from agentflow.utils.validators import ValidationError

raise ValidationError(
message="Prompt injection detected",
violation_type="prompt_injection",
details={"pattern": "ignore previous instructions"}
)

Fix: Sanitize user input, use input validators, enable strict mode for production.


Error Code Quick Reference

Code PrefixCategoryRetryableBase Class
GRAPH_000+Graph operationsNoGraphError
NODE_000+Node executionDependsNodeError
RECURSION_000Recursion limitsNoGraphRecursionError
STORAGE_000Generic storageNoStorageError
STORAGE_TRANSIENT_000+Retryable storageYesTransientStorageError
STORAGE_SERIALIZATION_000SerializationNoSerializationError
STORAGE_SCHEMA_000Schema/migrationNoSchemaVersionError
STORAGE_NOT_FOUND_000Resource not foundNoResourceNotFoundError
METRICS_000MetricsNoMetricsError
MEDIA_000Media inputNoUnsupportedMediaInputError
VALIDATION_000ValidationNoValidationError

Structured Error Responses

All AgentFlow errors include a to_dict() method for structured logging and API responses:

try:
await agent.process(message)
except GraphError as e:
error_response = e.to_dict()
# {
# "error_type": "GraphRecursionError",
# "error_code": "RECURSION_000",
# "message": "Recursion limit exceeded",
# "context": {"recursion_depth": 100, "max_depth": 50}
# }

Handling Errors

Basic Error Handling

from agentflow.core.exceptions import (
GraphError,
GraphRecursionError,
StorageError,
TransientStorageError,
)

try:
result = await agent.process(message)
except GraphRecursionError as e:
# Recursion limit exceeded - increase limit or fix loop
print(f"Recursion error: {e.error_code}")
except TransientStorageError as e:
# Retry with backoff
await retry_with_backoff(operation)
except StorageError as e:
# Non-retryable storage error
print(f"Storage error: {e.error_code}")
except GraphError as e:
# Generic graph error
print(f"Graph error: {e.error_code}")

Retry Logic for Transient Errors

import asyncio

async def retry_with_backoff(func, max_retries=3, base_delay=1.0):
for attempt in range(max_retries):
try:
return await func()
except TransientStorageError:
if attempt == max_retries - 1:
raise
delay = base_delay * (2 ** attempt)
await asyncio.sleep(delay)