OpenForgeAI

Open-source agentic architecture for production systems.

Built by one person. Running in production. Now open-source.

What Is This?

OpenForgeAI is the architecture framework extracted from Saarathi — a production AI-native platform built by a solo founder that competes with funded teams.

It provides the patterns you need to build systems where AI agents work as a real team: communicating via events, orchestrating multi-step workflows, and self-registering their capabilities.

This isn't a toy. These patterns run in production handling real users, real payments, and real-time messaging.

Core Components

EventBus — Agent Communication

A pub/sub event system where agents subscribe to events, emit new ones, and react autonomously. No direct function calls between agents.

from openforgeai.events import EventBus, Event

bus = EventBus()
bus.register_agent(my_agent)      # Agent subscribes to events it handles
await bus.publish(OrderCreated(order_id="123"))  # All subscribers react

Skill Registry — Agent Discovery

Auto-discovery and registration of agent skills. Drop a new skill file, it registers itself.

from openforgeai.agents import SkillRegistry

registry = SkillRegistry()
registry.discover("./skills/")    # Auto-finds all skills
registry.initialize(tenant_id="abc")  # Creates instances, wires EventBus

Saga Coordinators — Multi-Step Orchestration

Event-driven state machines for complex workflows. Each step completes on event arrival, not await calls.

from openforgeai.sagas import SagaCoordinator

class OnboardingSaga(SagaCoordinator):
    steps = ["send_welcome", "create_profile", "notify_team"]
    # Each step emits an event → next step triggers on completion event

Workflow Engine — Visual Process Automation

Define workflows as node graphs. Delay nodes, condition nodes, action nodes — all executing via the EventBus.

Deploy Validator — Pre-Deploy Safety

Catches broken imports, missing agent registrations, and compliance violations before you deploy.

python -m openforgeai.validators.deploy_check
# ✓ All skills imported
# ✓ All events have subscribers
# ✓ No orphan handlers

The 17 Laws of Agentic Engineering

The methodology behind the architecture. Read the full guide →

Quick Start

pip install openforgeai

from openforgeai import EventBus, BaseAgent, Event

# Define an event
class TaskCreated(Event):
    task_id: str
    title: str

# Define an agent
class NotificationAgent(BaseAgent):
    consumes_events = [TaskCreated]

    async def on_task_created(self, event: TaskCreated):
        print(f"New task: {event.title}")

# Wire it up
bus = EventBus()
agent = NotificationAgent()
bus.register_agent(agent)

await bus.publish(TaskCreated(task_id="1", title="Ship it"))
# Output: New task: Ship it

Who Is This For?

• Solo founders building production SaaS without a team
• Startup CTOs who want 10X team output with agentic patterns
• Senior engineers transitioning to AI-native architecture
• Anyone tired of AI demos that break in production

Learn More

• The 17 Laws of Agentic Engineering
• Architecture Guide
• CLAUDE.md Template — Drop this into any project
• Session Protocol — How to manage AI collaboration

The Story

I built Saarathi — an AI-native platform with real-time WhatsApp nurturing, automated webinar funnels, payment processing, CRM, and 14 AI agent skills working as a team. One person. Zero employees. Production-deployed. Real revenue.

The secret isn't "AI writes my code." The secret is architecture.

When your codebase is spaghetti, AI is a liability. When your codebase has clean contracts — EventBus, saga coordinators, skill registries — AI becomes a genuine team member.

Software engineering isn't dead. It's reinvented.

License

MIT — Use it, fork it, build with it.

Links

• Website: openforgeai.com
• GitHub: github.com/openforgeai
• Author: Goutam Biswas