FORGE

The Idea

LOGOS needs to orchestrate tens of thousands of AI agents programmatically. You can't do that by clicking in an IDE.

The solution: Anthropic's Claude Agent SDK - the same technology that powers Cursor's agent mode, but accessible from Python code.

Quick Start - Run In Your CLI

Step 1: Install the SDK

Step 2: Test the CLI Tool

Step 3: Invoke Your First Agent

Step 4: Try a Security Audit

Step 5: Check API Status (if Flask is running)

Architecture

┌─────────────────────────────────────────────────────────┐
│                    Your Python Code                      │
│         (Scripts, Flask routes, async jobs)              │
└────────────────────────┬────────────────────────────────┘
                         │
          ┌──────────────┼──────────────┐
          │              │              │
          ▼              ▼              ▼
    ┌──────────┐   ┌──────────┐   ┌──────────┐
    │   CLI    │   │ REST API │   │  Direct  │
    │  Tool    │   │ /api/    │   │  Import  │
    └────┬─────┘   └────┬─────┘   └────┬─────┘
         │              │              │
         └──────────────┼──────────────┘
                        │
                        ▼
              ┌─────────────────┐
              │  Bot Factory    │  ← Templates, fleet mgmt
              └────────┬────────┘
                       │
                       ▼
              ┌─────────────────┐
              │ Agent           │  ← Single invocation
              │ Orchestrator    │
              └────────┬────────┘
                       │
                       ▼
              ┌─────────────────┐
              │ Claude Agent    │  ← Anthropic SDK
              │ SDK             │
              └────────┬────────┘
                       │
                       ▼
              ┌─────────────────┐
              │ Tool Execution  │  ← Read, Write, Edit,
              │                 │     Bash, Grep, Glob
              └─────────────────┘

Code Guide

Direct Python Usage

# Single agent invocation
import asyncio
from winbusiness.Logos.Cortex.agent_orchestrator import AgentOrchestrator

async def main():
    orchestrator = AgentOrchestrator(
        default_working_directory="/root/Winbusiness"
    )
    
    result = await orchestrator.invoke_agent(
        task="Fix the bug in auth.py",
        personality="Sage",  # Uses LOGOS personality
        allowed_tools=["Read", "Write", "Edit", "Bash", "Grep", "Glob"],
    )
    
    print(f"Success: {result.success}")
    print(f"Output: {result.output}")
    print(f"Cost: ${result.total_cost_usd:.4f}")

asyncio.run(main())Copy

Bot Factory - Templates & Fleets

# Spawn a fleet of bots
import asyncio
from winbusiness.Logos.Cortex.bot_factory import BotFactory

async def main():
    factory = BotFactory(max_concurrent_bots=50)
    
    # Spawn 100 security auditor bots
    bots = await factory.spawn_fleet("SecurityAuditor", count=100)
    
    # Create tasks for each bot
    tasks = [
        {"bot_id": bot.bot_id, "task": f"Audit module_{i}.py"}
        for i, bot in enumerate(bots)
    ]
    
    # Execute in parallel
    results = await factory.parallel_execute(tasks)
    
    print(f"Completed: {sum(1 for r in results if r.success)}/{len(results)}")

asyncio.run(main())Copy

Available Bot Templates

REST API Reference

When Flask is running, these endpoints are available at /logos/api/agent/

Check Status

List Templates

Invoke Agent

curl -X POST http://localhost:5000/logos/api/agent/invoke \
  -H "Content-Type: application/json" \
  -d '{
    "prompt": "List all Python files in the project",
    "personality": "Default",
    "working_directory": "/root/Winbusiness"
  }'Copy

Batch Invoke

curl -X POST http://localhost:5000/logos/api/agent/batch \
  -H "Content-Type: application/json" \
  -d '{
    "tasks": [
      {"prompt": "Review auth.py", "personality": "Sage"},
      {"prompt": "Document api.py", "personality": "Quill"}
    ],
    "max_concurrent": 5
  }'Copy

Spawn Bot Fleet

curl -X POST http://localhost:5000/logos/api/fleet/spawn \
  -H "Content-Type: application/json" \
  -d '{
    "template_name": "BugHunter",
    "count": 10,
    "id_prefix": "hunter-"
  }'Copy

Complete Examples

Example 1: Security Audit Pipeline

#!/usr/bin/env python3
"""Security audit across multiple files"""
import asyncio
from winbusiness.Logos.Cortex.bot_factory import BotFactory

async def security_audit():
    factory = BotFactory()
    
    # Files to audit
    files = ["app.py", "config.py", "routes.py"]
    
    tasks = [
        {
            "template_name": "SecurityAuditor",
            "task": f"Perform a security audit of {f}. Look for: SQL injection, XSS, hardcoded secrets, improper input validation."
        }
        for f in files
    ]
    
    results = await factory.parallel_execute(tasks, max_concurrent=3)
    
    # Compile report
    print("=" * 60)
    print("SECURITY AUDIT REPORT")
    print("=" * 60)
    for i, result in enumerate(results):
        print(f"\n--- {files[i]} ---")
        print(result.output[:500] if result.output else "No issues found")
    
asyncio.run(security_audit())Copy

Example 2: Documentation Generator

#!/usr/bin/env python3
"""Generate documentation for all Python files in a directory"""
import asyncio
from winbusiness.Logos.Cortex.agent_orchestrator import AgentOrchestrator

async def generate_docs():
    orchestrator = AgentOrchestrator(
        default_working_directory="/root/Winbusiness"
    )
    
    result = await orchestrator.invoke_agent(
        task="""
        Find all Python files in winbusiness/Logos/Cortex/ and create
        a documentation summary for each one. Include:
        - Purpose of the file
        - Main classes and functions
        - Usage examples
        
        Save the documentation to a file called API_DOCS.md
        """,
        personality="Quill",  # Documentation specialist
    )
    
    print(result.output)

asyncio.run(generate_docs())Copy

Example 3: Interactive Session

This starts an interactive session where you can have a multi-turn conversation with an agent. The agent remembers context between messages.