Penguin Python API Reference
This page documents the public APIs that ship today. Anything not listed here is work-in-progress and tracked in the future considerations roadmap.
Installation
pip install penguin-ai # includes CLI and Python package
Quick-start
from penguin.agent import PenguinAgent
agent = PenguinAgent()
print(agent.chat("Hello Penguin!"))
Modules & Classes
| Import path | Exists? | Notes |
|---|---|---|
penguin.agent.PenguinAgent | ✅ | Synchronous wrapper around PenguinCore – chat, stream, run_task |
penguin.agent.PenguinAgentAsync | ✅ | Async variant with identical method names (returning coroutines) |
penguin.project.manager.ProjectManager | ✅ | SQLite-backed project CRUD and basic task helpers |
penguin.web.app.PenguinAPI | ✅ | Programmatic wrapper aligned with current RunMode/task clarification truth |
penguin.core.PenguinCore | ✅ | Low-level runtime object; advanced/compatibility surface |
penguin.tools.ToolManager | ✅ | Runtime registry & execution of tools |
Anything else you may have seen in earlier drafts (Memory providers, BatchProcessor, PerformanceMonitor, custom plugin framework, etc.) has not landed yet.
PenguinAgent
from penguin.agent import PenguinAgent
agent = PenguinAgent()
Methods
| Method | Description |
|---|---|
| `chat(message: str, *, context: dict | None = None) -> str` |
| `stream(message: str, *, context: dict | None = None) -> Iterator[str]` |
run_task(prompt: str, *, max_iterations: int | None = None) -> dict | Multi-step reasoning/action loop using core.run_mode; a limit is applied only when supplied. |
new_conversation() -> str | Start fresh conversation, returns session id. |
load_conversation(session_id: str) -> bool | Load a saved session into memory. |
All other attributes or methods are considered internal and may change without notice.
Example
conv = agent.new_conversation()
agent.chat("Explain asyncio in Python", context={"conversation_id": conv})
PenguinAgentAsync
from penguin.agent import PenguinAgentAsync
agent = await PenguinAgentAsync.create()
- Same public surface as
PenguinAgent, but every method isasyncand returns an awaitable.
PenguinAPI
from penguin.web.app import PenguinAPI
api = PenguinAPI()
Methods
| Method | Description |
|---|---|
chat(...) -> dict | Chat through the same backend core used by the web surface. |
| `run_task(task_description: str, max_iterations: int | None = None, project_id: str |
| `resume_with_clarification(task_id: str, answer: str, answered_by: str | None = None) -> dict` |
This is the current lightweight Python embedding surface for clarification-aware task execution. A deeper audit/ergonomics pass is intentionally tracked separately because this surface is still thinner and less polished than the CLI and web/API paths.
ProjectManager
Basic project / task operations. ProjectManager is the current documented entry point; earlier references to AsyncProjectManager were premature and should not be treated as shipped public API.
from pathlib import Path
from penguin.project.manager import ProjectManager
from penguin.project.models import TaskStatus
pm = ProjectManager(Path("./penguin-workspace"))
project = pm.create_project(name="Demo", description="Example project")
print("Project id:", project.id)
# Tasks
task = pm.create_task(
title="Initial research",
description="Map the repository and identify entry points",
project_id=project.id,
)
pm.update_task_status(task.id, TaskStatus.ACTIVE)
pm.update_task_status(task.id, TaskStatus.COMPLETED)
Implemented high-level methods:
create_project(name, description="", ...)list_projects(status: str | None = None)delete_project(project_id)create_task(title, description, project_id=None, parent_task_id=None, priority=0, ...)list_tasks(project_id: str | None = None, status: TaskStatus | None = None)update_task_status(task_id, status: TaskStatus)delete_task(task_id)
Dependency graphs, Blueprint diagnostics, typed dependency policies, and clarification-aware execution exist deeper in the runtime/backend, but the Python library surface for them is intentionally narrower today. Broader library-surface polish is tracked separately.
PenguinCore (advanced)
For power-users that need direct access to the orchestration layer.
from penguin.core import PenguinCore
core = await PenguinCore.create(enable_cli=False)
resp = await core.process("Summarise this repository in 3 points")
print(resp["assistant_response"])
Key async methods you can rely on:
process(input_data, *, streaming=False, stream_callback=None)start_run_mode(name, description=None, continuous=False, time_limit=None)run_session_goal(session_id, *, max_iterations=None, timeout_seconds=None, directory=None)
run_session_goal() executes an already-persisted active session goal and
returns its updated goal state plus the RunMode result. It does not create a
goal. Create and control the durable goal through the session-goal HTTP surface
or TUI first.
directory, when supplied, is a scope assertion. For a legacy session with no
stored directory, the first valid explicit directory is durably bound as its
root; after that it must resolve to the same persisted directory.
result = await core.run_session_goal(
"session_20260709_193237_076f3c0d",
max_iterations=8,
timeout_seconds=600,
)
print(result["status"])
print(result["goal"]["objective"])
The target must be a saved session with an active goal. When omitted,
max_iterations, timeout_seconds, and the goal's cumulative token_budget
remain None; Penguin supplies no local default or hard maximum. Positive
values explicitly supplied by the caller are enforced. Usage is known only
after a provider response, so the last billed turn can cross an explicitly
configured threshold before the runtime refuses another turn.
Missing state, lifecycle conflicts, and validation failures are exposed as
typed goal exceptions at this low-level surface and as 404, 409, and 422
through HTTP.
PenguinCore primarily handles construction, delegation, and compatibility
methods. Runtime behavior is owned by modules under penguin.core_runtime,
penguin.engine,
penguin.run_mode, and related service/domain packages. Prefer higher-level
client APIs unless you specifically need this compatibility surface.
ToolManager
from penguin.tools import ToolManager
manager = ToolManager()
print([t.name for t in manager.list_tools()])
You can register custom tools via the decorator:
@manager.register("my_tool")
def my_tool(**kwargs):
return {"echo": kwargs}
Deprecated / Future APIs
The following names appeared in earlier documentation but do not exist in the current public release. They are tracked on the roadmap and should not be imported yet:
BatchProcessorPerformanceMonitorErrorRecoveryPlugin/plugin_hookMemoryProvidersubclasses beyond default SQLite providerAgentBuilder
See the "Python API Roadmap" section in future considerations for planned timelines.
Last updated: May 27th 2026