Skip to content

Python SDK

Programmatic access to Nova AI capabilities.

Installation

# Install from source
git clone https://github.com/Jaureguy760/nova_ai.git
cd nova_ai
pip install -e .

Quick Start

from pathlib import Path
from src.orchestrator.claude_sdk_executor import ClaudeSDKExecutor

# Create executor
executor = ClaudeSDKExecutor(
    project_root=Path.cwd(),
    agent_name="orchestrator",
    use_sdk_mcp=True
)

# Run a task
result = await executor.run_task("implement user authentication")
print(result.modified_files)  # list[Path]

ClaudeSDKExecutor

Main orchestration interface.

Constructor

executor = ClaudeSDKExecutor(
    project_root: Path,           # Project root directory
    agent_name: str = "orchestrator",  # Agent to use
    use_sdk_mcp: bool = True,     # Use in-process MCP
    session_id: str = None,       # Resume existing session
    model: str = None             # Override model
)

Methods

run_task()

Execute a task with the configured agent.

result = await executor.run_task(
    task: str,                    # Natural language task
    max_turns: int = 50,          # Maximum agent turns
    timeout: int = 600            # Timeout in seconds
)

Returns: ExecutorResult

from src.orchestrator.executor.result import ExecutorResult

@dataclass(frozen=True)
class ExecutorResult:
    transcript: str                              # Full conversation transcript
    summary: str                                 # Brief summary of accomplishment
    verification: VerificationResult | None     # Quality gate results
    modified_files: list[Path] | None           # Files that were modified
    session_id: str | None                       # Session ID for continuation
    total_cost_usd: float                        # Cost tracking

    @property
    def success(self) -> bool:                   # Derived from verification
        ...

continue_session()

Continue from a previous session.

# First task
result1 = await executor.run_task("create user model")

# Continue in same session
executor2 = ClaudeSDKExecutor(
    project_root=Path.cwd(),
    session_id=result1.session_id
)
result2 = await executor2.run_task("add validation")

Knowledge Base

Search the 5.4GB vector database.

from src.orchestrator.tools.kb_tools import search_knowledge_base

# Synchronous search function
results = search_knowledge_base(
    query="JWT authentication patterns",
    top_k=5,
    kb_dir=None  # Uses default vector/local-kb
)

if results["status"] == "ok":
    for result in results["results"]:
        print(f"{result['path']}: {result['score']}")
        print(result['chunk'][:200])

GitHub Operations

Interact with GitHub via MCP tools.

from src.orchestrator.tools.github_tools import (
    create_pr_tool,
    list_issues_tool,
    list_prs_tool,
    update_issue_tool
)

# Create PR (MCP tool format - takes args dict)
result = await create_pr_tool({
    "title": "Add authentication",
    "body": "Implements JWT auth flow",
    "base": "main",
    "head": "feature/auth"
})

# List issues
issues = await list_issues_tool({
    "state": "open",
    "labels": "bug"
})

# Update issue
await update_issue_tool({
    "issue_number": 123,
    "state": "closed"
})

Code Analysis

Use Blarify for code intelligence.

from src.orchestrator.tools.blarify_tools import (
    analyze_dependencies_tool,
    check_circular_deps_tool,
)

# Check for circular imports (MCP tool format)
result = await check_circular_deps_tool({
    "entity_id": "nova_ai",  # Project identifier
    "scope": "src/"
})
if result.get("cycles"):
    print(f"Found {len(result['cycles'])} circular dependencies")

# Map module dependencies
graph = await analyze_dependencies_tool({
    "entity_id": "nova_ai",
    "path": "src/auth/"
})
print(graph.get("summary", ""))

Memory

Persistent storage across sessions.

from src.orchestrator.tools.memory_tools import (
    get_memory_sdk_mcp_server,
    NamespacedMemoryStore,
    get_namespaced_store,
)

# Use namespaced memory for cross-session storage
store = get_namespaced_store()

# Store data in a namespace
await store.store("project:nova_ai", "context", {
    "framework": "FastAPI",
    "database": "PostgreSQL"
})

# Retrieve later
context = await store.retrieve("project:nova_ai", "context")

# List keys in namespace
keys = await store.list_keys("project:nova_ai")

Parallel Execution

Run multiple agents concurrently.

from src.orchestrator.parallel_executor import ParallelSandboxedExecutor

# Define independent tasks
tasks = [
    {"task": "fix test_auth.py", "agent": "debugger"},
    {"task": "fix test_user.py", "agent": "debugger"},
    {"task": "fix test_payment.py", "agent": "debugger"}
]

# Execute in parallel (3-5x speedup)
executor = ParallelSandboxedExecutor(tasks)
results = await executor.run_all()

Headless Mode

For CI/CD pipelines.

from pathlib import Path
from src.orchestrator.background_sdk_executor_pkg.executor import BackgroundSDKExecutor

# Create executor for headless operation
executor = BackgroundSDKExecutor(project_root=Path.cwd())

# Run task in headless mode (no interactive UI)
result = await executor.run_headless_cli(
    task="review code for security issues",
    agent_name="code-reviewer",
    timeout_seconds=600
)

# Or use the CLI script directly:
# python scripts/nova_headless.py "review code" --output-format json

Error Handling

from src.orchestrator.errors import (
    AgentError,
    MCPToolError,
    ScriptError,
)

try:
    result = await executor.run_task("implement feature")
except AgentError as e:
    print(f"Agent failed: {e}")
except MCPToolError as e:
    print(f"MCP tool error: {e}")
except ScriptError as e:
    print(f"Script error: {e}")
except TimeoutError:  # Built-in Python exception
    print("Task timed out")

What's Next?