Forge documentation
Library referencePython

forge.core

Python core package; imports are explicit from this package.

Python core package; imports are explicit from this package.

Package contract

FieldValue
Languagepython
Source version0.1.0
Manifestforge-py/pyproject.toml
Source files16
EvidenceSource reference; registry publication and runtime conformance are separate checks

Import boundary

import forge.core

Use a source checkout or your verified private registry. Manifest coordinates identify the package; they do not establish that a public registry release exists.

Source reference

Download package reference JSON. Each original source file and generated declaration artifact has its own SHA-256 digest. Function bodies and constant values are omitted from downloads. These are source declaration inventories, not compiler-resolved rustdoc, TypeDoc, DocC, or Dokka output. Private modules can contain public declarations that are not reachable through the package boundary; consult the entry point before importing.

brew.py

Read declaration text · 34 declaration entries

class JoinMode(Enum)

class JoinModeFirstN()

def to_dict(self) -> dict[str, Any]

def join_mode_to_dict(mode: JoinMode | JoinModeFirstN) -> Any

def join_mode_from_dict(data: Any) -> JoinMode | JoinModeFirstN

class BrewId()

def as_str(self) -> str

class NodeId()

def as_str(self) -> str

class BrewVersion()

def as_str(self) -> str

class BrewEdgeKind(Enum)

class BrewEdge()

def to_dict(self) -> dict[str, Any]

def from_dict(data: dict[str, Any]) -> BrewEdge

def sort_key(self) -> tuple[str, str, str]

class AgentStepNode()

class ToolInvocationNode()

class McpCallNode()

class WebOperationNode()

class ConditionalBranchNode()

class ParallelForkNode()

class SubBrewRefNode()

class HumanCheckpointNode()

def brew_node_kind_to_dict(kind: BrewNodeKind) -> dict[str, Any]

def brew_node_kind_from_dict(data: dict[str, Any]) -> BrewNodeKind

class BrewNode()

def to_dict(self) -> dict[str, Any]

def from_dict(data: dict[str, Any]) -> BrewNode

class Brew()

def to_dict(self) -> dict[str, Any]

def from_dict(data: dict[str, Any]) -> Brew

def to_json(self) -> str

def from_json(data: str) -> Brew

brew_builder.py

Read declaration text · 10 declaration entries

class BrewBuilder()

def __init__(self, brew_id: str, version: str) -> None

def add_node(self, node_id: str, kind: BrewNodeKind) -> BrewBuilder

def add_edge(self, from_id: str, to_id: str, edge_kind: BrewEdgeKind) -> BrewBuilder

def set_entry(self, node_id: str) -> BrewBuilder

def set_exit(self, node_id: str) -> BrewBuilder

def with_topology(self, topology: ModelTopology) -> BrewBuilder

def with_metadata(self, node_id: str, key: str, value: str) -> BrewBuilder

def build(self) -> Brew

def topological_sort(nodes: dict[str, BrewNode], edges: list[BrewEdge]) -> list[str]

brew_resolver.py

Read declaration text · 11 declaration entries

class BrewResolutionErrorKind(Enum)

class BrewResolutionError()

def to_dict(self) -> dict[str, Any]

class BrewEnvironment()

def to_dict(self) -> dict[str, Any]

def compute_hash(self) -> str

class ResolvedBrewPlan()

def to_dict(self) -> dict[str, Any]

class BrewResolutionFailed(Exception)

def __init__(self, errors: list[BrewResolutionError]) -> None

def resolve(brew: Brew, env: BrewEnvironment) -> ResolvedBrewPlan

config.py

Read declaration text · 18 declaration entries

class GenerateOptions()

def with_temperature(self, temperature: float) -> GenerateOptions

def with_max_tokens(self, max_tokens: int) -> GenerateOptions

def with_top_p(self, top_p: float) -> GenerateOptions

def with_stop_sequences(self, sequences: list[str]) -> GenerateOptions

def with_frequency_penalty(self, penalty: float) -> GenerateOptions

def with_presence_penalty(self, penalty: float) -> GenerateOptions

def with_seed(self, seed: int) -> GenerateOptions

def with_output_schema(self, schema: JsonSchema) -> GenerateOptions

def to_dict(self) -> dict[str, Any]

def from_dict(data: dict[str, Any]) -> GenerateOptions

def to_json(self) -> str

def from_json(data: str) -> GenerateOptions

class EmbedOptions()

def with_model(self, model: str) -> EmbedOptions

def with_dimensions(self, dimensions: int) -> EmbedOptions

def to_dict(self) -> dict[str, Any]

def from_dict(data: dict[str, Any]) -> EmbedOptions

error.py

Read declaration text · 39 declaration entries

class ForgeError(Exception)

class ForgeProviderNotFoundError(ForgeError)

def __init__(self, provider_ref: str) -> None

class ForgeInvalidProviderRefError(ForgeError)

def __init__(self, input_str: str) -> None

class ForgeSchemaValidationError(ForgeError)

def __init__(self, path: str, reason: str) -> None

class InvalidLifecycleTransitionError(ForgeError)

def __init__(self, from_state: str, to_state: str, reason: str) -> None

class ForgeToolInvocationDeniedError(ForgeError)

def __init__(self, tool_name: str, agent_did: str, required_capability: str, act_id: str) -> None

class ForgeToolExecutionError(ForgeError)

def __init__(self, tool_name: str, reason: str) -> None

class ForgeJsonError(ForgeError)

def __init__(self, message: str) -> None

class ForgeMissingConfigError(ForgeError)

def __init__(self, field: str, hint: str) -> None

class ForgeStopConditionError(ForgeError)

def __init__(self, reason: str, steps: int, tokens: int) -> None

class ForgeTelemetryError(ForgeError)

def __init__(self, reason: str) -> None

class ForgeUnsupportedOperationError(ForgeError)

def __init__(self, model: str, operation: str) -> None

class ForgeProviderUnavailableError(ForgeError)

def __init__(self, provider_ref: str, reason: str) -> None

class ForgeProviderAuthenticationFailedError(ForgeError)

def __init__(self, provider_ref: str, reason: str) -> None

class ForgeCapabilityUnsupportedError(ForgeError)

def __init__(self, provider_ref: str, capability: str) -> None

class ForgeProviderNegotiationFailedError(ForgeError)

def __init__(self, provider_ref: str, reason: str) -> None

class ForgeProviderSessionExpiredError(ForgeError)

def __init__(self, provider_ref: str, session_id: str) -> None

class ForgeProviderInterruptUnsupportedError(ForgeError)

def __init__(self, provider_ref: str) -> None

class ForgeProviderResumeUnsupportedError(ForgeError)

def __init__(self, provider_ref: str) -> None

class ForgeInternalError(ForgeError)

def __init__(self, message: str) -> None

message.py

Read declaration text · 15 declaration entries

class Role(Enum)

def as_str(self) -> str

class TextPart()

class ImagePart()

class ToolCallPart()

class ToolResultPart()

class ModelMessage()

def new(role: Role, parts: list[MessagePart]) -> ModelMessage

def text(role: Role, text: str) -> ModelMessage

def tool_calls(self) -> list[ToolCallPart]

def text_content(self) -> str

def to_dict(self) -> dict[str, Any]

def from_dict(data: dict[str, Any]) -> ModelMessage

def to_json(self) -> str

def from_json(data: str) -> ModelMessage

model.py

Read declaration text · 9 declaration entries

class LanguageModel(ABC)

def model_id(self) -> str

def provider(self) -> str

async def generate(self, messages: list[object], tools: list[ToolDefinition], options: GenerateOptions) -> GenerateResult

async def stream(self, messages: list[object], tools: list[ToolDefinition], options: GenerateOptions) -> AsyncGenerator[StreamChunk, None]

def supports_tool_calling(self) -> bool

def supports_structured_output(self) -> bool

def supports_image_input(self) -> bool

def supports_streaming(self) -> bool

output.py

Read declaration text · 29 declaration entries

class FinishReason(Enum)

def is_complete(self) -> bool

def is_tool_call(self) -> bool

class Usage()

def zero() -> Usage

def add(self, other: Usage) -> Usage

def to_dict(self) -> dict[str, int]

def from_dict(data: dict[str, Any]) -> Usage

def to_json(self) -> str

def from_json(data: str) -> Usage

class GenerateResult()

def text(self) -> str

def has_tool_calls(self) -> bool

def to_dict(self) -> dict[str, Any]

def from_dict(data: dict[str, Any]) -> GenerateResult

def to_json(self) -> str

def from_json(data: str) -> GenerateResult

class TextDelta()

def as_text(self) -> str | None

class ToolCallDelta()

def as_text(self) -> str | None

class StreamDone()

def as_text(self) -> str | None

def stream_chunk_text_delta(text: str) -> TextDelta

def stream_chunk_done(finish_reason: FinishReason, usage: Usage) -> StreamDone

def is_text_delta(chunk: StreamChunk) -> bool

def is_done(chunk: StreamChunk) -> bool

def stream_chunk_to_dict(chunk: StreamChunk) -> dict[str, Any]

def stream_chunk_from_dict(data: dict[str, Any]) -> StreamChunk

provider.py

Read declaration text · 55 declaration entries

class ProviderRef()

def parse(input_str: str) -> ProviderRef

def as_str(self) -> str

def to_json(self) -> str

def from_json(data: str) -> ProviderRef

class ProviderMetadata()

def to_dict(self) -> dict[str, Any]

class RuntimeCapability(str, Enum)

def ordinal(self) -> int

class ProviderRuntimeCapabilities()

def baseline() -> ProviderRuntimeCapabilities

def with_capability(self, capability: RuntimeCapability) -> ProviderRuntimeCapabilities

def supports(self, capability: RuntimeCapability) -> bool

def to_dict(self) -> dict[str, Any]

class ProviderNegotiationRequest()

class ProviderNegotiationResult()

class ProviderSessionState(str, Enum)

class ProviderUsageSummary()

class ProviderSessionEvent()

class ProviderRegistry()

def __init__(self) -> None

def register(self, provider_ref: str, model: LanguageModel) -> None

def register_with_runtime(self, provider_ref: str, model: LanguageModel, runtime_capabilities: ProviderRuntimeCapabilities) -> None

def get(self, provider_ref: str) -> LanguageModel | None

def require(self, provider_ref: str) -> LanguageModel

def metadata(self, provider_ref: str) -> ProviderMetadata | None

def list(self) -> list[str]

def is_empty(self) -> bool

def negotiate(self, request: ProviderNegotiationRequest) -> ProviderNegotiationResult

class ProviderFamily(str, Enum)

class AuthStrategy(str, Enum)

class ProviderPreset()

class ProviderExecutionRequest()

class ProviderInstallOptions()

def approved_coding_provider_presets() -> list[ProviderPreset]

def approved_direct_provider_presets() -> list[ProviderPreset]

def approved_gateway_provider_presets() -> list[ProviderPreset]

def register_official_coding_providers(registry: ProviderRegistry, *, env: ProviderEnvironment | None=None, generate: ProviderGenerateHook | None=None, stream: ProviderStreamHook | None=None) -> None

def register_default_core_direct_providers(registry: ProviderRegistry, *, env: ProviderEnvironment | None=None, generate: ProviderGenerateHook | None=None, stream: ProviderStreamHook | None=None) -> list[str]

def register_openai_model(registry: ProviderRegistry, model_id: str, *, env: ProviderEnvironment | None=None, generate: ProviderGenerateHook | None=None, stream: ProviderStreamHook | None=None) -> str

def register_anthropic_model(registry: ProviderRegistry, model_id: str, *, env: ProviderEnvironment | None=None, generate: ProviderGenerateHook | None=None, stream: ProviderStreamHook | None=None) -> str

def register_google_model(registry: ProviderRegistry, model_id: str, *, env: ProviderEnvironment | None=None, generate: ProviderGenerateHook | None=None, stream: ProviderStreamHook | None=None) -> str

def register_xai_model(registry: ProviderRegistry, model_id: str, *, env: ProviderEnvironment | None=None, generate: ProviderGenerateHook | None=None, stream: ProviderStreamHook | None=None) -> str

def register_deepseek_model(registry: ProviderRegistry, model_id: str, *, env: ProviderEnvironment | None=None, generate: ProviderGenerateHook | None=None, stream: ProviderStreamHook | None=None) -> str

def register_mistral_model(registry: ProviderRegistry, model_id: str, *, env: ProviderEnvironment | None=None, generate: ProviderGenerateHook | None=None, stream: ProviderStreamHook | None=None) -> str

def register_cohere_model(registry: ProviderRegistry, model_id: str, *, env: ProviderEnvironment | None=None, generate: ProviderGenerateHook | None=None, stream: ProviderStreamHook | None=None) -> str

def register_groq_model(registry: ProviderRegistry, model_id: str, *, env: ProviderEnvironment | None=None, generate: ProviderGenerateHook | None=None, stream: ProviderStreamHook | None=None) -> str

def register_moonshot_model(registry: ProviderRegistry, model_id: str, *, env: ProviderEnvironment | None=None, generate: ProviderGenerateHook | None=None, stream: ProviderStreamHook | None=None) -> str

def register_zai_model(registry: ProviderRegistry, model_id: str, *, env: ProviderEnvironment | None=None, generate: ProviderGenerateHook | None=None, stream: ProviderStreamHook | None=None) -> str

def register_minimax_model(registry: ProviderRegistry, model_id: str, *, env: ProviderEnvironment | None=None, generate: ProviderGenerateHook | None=None, stream: ProviderStreamHook | None=None) -> str

def register_openrouter_model(registry: ProviderRegistry, model_id: str, *, env: ProviderEnvironment | None=None, generate: ProviderGenerateHook | None=None, stream: ProviderStreamHook | None=None) -> str

def register_bedrock_model(registry: ProviderRegistry, model_id: str, *, env: ProviderEnvironment | None=None, generate: ProviderGenerateHook | None=None, stream: ProviderStreamHook | None=None) -> str

def register_vertex_ai_model(registry: ProviderRegistry, model_id: str, *, env: ProviderEnvironment | None=None, generate: ProviderGenerateHook | None=None, stream: ProviderStreamHook | None=None) -> str

def register_microsoft_foundry_model(registry: ProviderRegistry, model_id: str, *, env: ProviderEnvironment | None=None, generate: ProviderGenerateHook | None=None, stream: ProviderStreamHook | None=None) -> str

def register_foundry_model(registry: ProviderRegistry, model_id: str, *, env: ProviderEnvironment | None=None, generate: ProviderGenerateHook | None=None, stream: ProviderStreamHook | None=None) -> str

routing.py

Read declaration text · 20 declaration entries

class TaskMode(Enum)

def as_role_name(self) -> str

class ExecutionTopology(Enum)

class ModelCapabilities()

def to_dict(self) -> dict[str, Any]

def from_dict(data: dict[str, Any]) -> ModelCapabilities

class RoutingContext()

def to_dict(self) -> dict[str, Any]

def from_dict(data: dict[str, Any]) -> RoutingContext

def to_json(self) -> str

def from_json(data: str) -> RoutingContext

class ResolvedRoute()

def to_dict(self) -> dict[str, Any]

def from_dict(data: dict[str, Any]) -> ResolvedRoute

class ModelRouter(ABC)

def name(self) -> str

def route(self, context: RoutingContext, topology: ModelTopology) -> ResolvedRoute

class DefaultModelRouter(ModelRouter)

def name(self) -> str

def route(self, context: RoutingContext, topology: ModelTopology) -> ResolvedRoute

schema.py

Read declaration text · 24 declaration entries

class SchemaType(Enum)

class JsonSchema()

def string() -> JsonSchema

def number() -> JsonSchema

def integer() -> JsonSchema

def boolean() -> JsonSchema

def array() -> JsonSchema

def object_() -> JsonSchema

def null() -> JsonSchema

def set_description(self, desc: str) -> JsonSchema

def property(self, name: str, schema: JsonSchema) -> JsonSchema

def required_(self, name: str) -> JsonSchema

def items_schema(self, schema: JsonSchema) -> JsonSchema

def set_enum_values(self, values: list[Any]) -> JsonSchema

def set_minimum(self, min_val: float) -> JsonSchema

def set_maximum(self, max_val: float) -> JsonSchema

def set_min_length(self, length: int) -> JsonSchema

def set_max_length(self, length: int) -> JsonSchema

def set_additional_properties(self, allowed: bool) -> JsonSchema

def validate(self, value: Any) -> None

def to_dict(self) -> dict[str, Any]

def from_dict(data: dict[str, Any]) -> JsonSchema

def to_json(self) -> str

def from_json(data: str) -> JsonSchema

telemetry.py

Read declaration text · 27 declaration entries

class SpanAttributeString()

class SpanAttributeInt()

class SpanAttributeFloat()

class SpanAttributeBool()

def to_span_attribute(value: str | int | float | bool) -> SpanAttribute

def span_attribute_value(attr: SpanAttribute) -> str | int | float | bool

class SpanStatusCode(Enum)

class SpanStatusUnset()

class SpanStatusOk()

class SpanStatusError()

class ForgeSpan()

def set_attribute(self, key: str, value: str | int | float | bool) -> None

def end(self) -> None

def end_with_error(self, message: str) -> None

def with_parent(self, parent_id: str) -> ForgeSpan

class ForgeEvent()

def set_attribute(self, key: str, value: str | int | float | bool) -> None

class TelemetryEmitter(ABC)

def emit_span(self, span: ForgeSpan) -> None

def emit_event(self, event: ForgeEvent) -> None

class NoopEmitter(TelemetryEmitter)

def emit_span(self, span: ForgeSpan) -> None

def emit_event(self, event: ForgeEvent) -> None

class RecordingEmitter(TelemetryEmitter)

def __init__(self) -> None

def emit_span(self, span: ForgeSpan) -> None

def emit_event(self, event: ForgeEvent) -> None

tool.py

Read declaration text · 30 declaration entries

class ToolTier(Enum)

def requires_authorization(self) -> bool

def as_str(self) -> str

class ToolDefinition()

def builder(name: str) -> ToolDefinitionBuilder

def to_dict(self) -> dict[str, Any]

def from_dict(data: dict[str, Any]) -> ToolDefinition

def to_json(self) -> str

def from_json(data: str) -> ToolDefinition

class ToolDefinitionBuilder()

def __init__(self, name: str) -> None

def description(self, desc: str) -> ToolDefinitionBuilder

def parameters(self, schema: JsonSchema) -> ToolDefinitionBuilder

def tier(self, tier: ToolTier) -> ToolDefinitionBuilder

def build(self) -> ToolDefinition

class ToolCall()

def to_dict(self) -> dict[str, Any]

def from_dict(data: dict[str, Any]) -> ToolCall

def to_json(self) -> str

def from_json(data: str) -> ToolCall

class ToolResult()

def to_dict(self) -> dict[str, Any]

def from_dict(data: dict[str, Any]) -> ToolResult

def to_json(self) -> str

def from_json(data: str) -> ToolResult

class ToolApprovalApprove()

class ToolApprovalDeny()

class ToolApprovalModify()

def tool_approval_to_dict(approval: ToolApproval) -> dict[str, Any]

def tool_approval_from_dict(data: dict[str, Any]) -> ToolApproval

topology.py

Read declaration text · 29 declaration entries

class CostPreference(Enum)

def as_str(self) -> str

class LatencyPreference(Enum)

def as_str(self) -> str

class ModelSlot()

def to_dict(self) -> dict[str, Any]

def from_dict(data: dict[str, Any]) -> ModelSlot

class ModelTopology()

def single(provider_ref: ProviderRef) -> ModelTopology

def builder() -> TopologyBuilder

def default_slot(self) -> ModelSlot

def slot_for_role(self, role: str) -> ModelSlot

def has_role(self, role: str) -> bool

def slot_count(self) -> int

def all_provider_refs(self) -> list[ProviderRef]

def to_dict(self) -> dict[str, Any]

def from_dict(data: dict[str, Any]) -> ModelTopology

def to_json(self) -> str

def from_json(data: str) -> ModelTopology

class TopologyBuilder()

def __init__(self) -> None

def name(self, name: str) -> TopologyBuilder

def slot(self, role: str, primary: ProviderRef) -> TopologyBuilder

def with_fallback(self, role: str, fallback: ProviderRef) -> TopologyBuilder

def with_required_capability(self, role: str, cap: RuntimeCapability) -> TopologyBuilder

def with_cost_preference(self, role: str, pref: CostPreference) -> TopologyBuilder

def with_latency_preference(self, role: str, pref: LatencyPreference) -> TopologyBuilder

def with_scope_narrowing(self, role: str, scopes: list[str]) -> TopologyBuilder

def build(self) -> ModelTopology

types.py

Read declaration text · 12 declaration entries

class AgentDid()

def new(did: str) -> AgentDid | None

def from_trusted(did: str) -> AgentDid

def as_str(self) -> str

def to_json(self) -> str

def from_json(data: str) -> AgentDid

class Timestamp()

def now() -> Timestamp

def from_iso8601(s: str) -> Timestamp | None

def to_iso8601(self) -> str

def to_json(self) -> str

def from_json(data: str) -> Timestamp | None

Continue

forge.comm

Python comm package; imports are explicit from this package.

forge.embed

Python embed package; imports are explicit from this package.

On this page

Package contractImport boundarySource referencebrew.pybrew_builder.pybrew_resolver.pyconfig.pyerror.pymessage.pymodel.pyoutput.pyprovider.pyrouting.pyschema.pytelemetry.pytool.pytopology.pytypes.pyContinue