Client API Reference

OpenIntentClient

The synchronous client for interacting with an OpenIntent server.

OpenIntentClient

OpenIntentClient(base_url: str, api_key: str, agent_id: str, timeout: float = 30.0)

Synchronous client for the OpenIntent Coordination Protocol.

Example

client = OpenIntentClient(
    base_url="https://api.openintent.ai",
    api_key="your-api-key",
    agent_id="my-agent"
)

# Create an intent
intent = client.create_intent(
    title="Research market trends",
    description="Analyze Q4 market data and identify patterns"
)

# Update state with optimistic concurrency
client.update_state(intent.id, intent.version, {"progress": 0.5})

# Acquire a lease for exclusive scope access
with client.lease(intent.id, "analysis") as lease:
    # Perform work within the leased scope
    client.log_event(intent.id, EventType.COMMENT, {"note": "Starting analysis"})

create_intent

create_intent(title: str, description: str = '', constraints: Optional[dict[str, Any]] = None, initial_state: Optional[dict[str, Any]] = None, parent_intent_id: Optional[str] = None, depends_on: Optional[list[str]] = None, governance_policy: Optional[dict[str, Any]] = None) -> Intent

Create a new intent.

Parameters:

Name	Type	Description	Default
title	str	Human-readable title for the intent.	required
description	str	Detailed description of the goal.	''
constraints	Optional[dict[str, Any]]	Optional constraints dictionary (e.g. {"rules": [...]}).	None
initial_state	Optional[dict[str, Any]]	Optional initial state data.	None
parent_intent_id	Optional[str]	Optional parent intent ID for hierarchical graphs (RFC-0002).	None
depends_on	Optional[list[str]]	Optional list of intent IDs this depends on (RFC-0002).	None
governance_policy	Optional[dict[str, Any]]	Optional server-enforced governance policy (RFC-0013). Supports: completion_mode, write_scope, allowed_agents, max_cost, quorum_threshold, require_status_reason.	None

Returns:

Type	Description
Intent	The created Intent object.

get_intent

get_intent(intent_id: str) -> Intent

Retrieve an intent by ID.

Parameters:

Name	Type	Description	Default
intent_id	str	The unique identifier of the intent.	required

Returns:

Type	Description
Intent	The Intent object.

list_intents

list_intents(status: Optional[IntentStatus] = None, limit: int = 50, offset: int = 0) -> list[Intent]

List intents with optional filtering.

Parameters:

Name	Type	Description	Default
status	Optional[IntentStatus]	Filter by intent status.	None
limit	int	Maximum number of results.	50
offset	int	Pagination offset.	0

Returns:

Type	Description
list[Intent]	List of Intent objects.

create_child_intent

create_child_intent(parent_id: str, title: str, description: str = '', constraints: Optional[dict[str, Any]] = None, initial_state: Optional[dict[str, Any]] = None, depends_on: Optional[list[str]] = None) -> Intent

Create a child intent under a parent intent.

Parameters:

Name	Type	Description	Default
parent_id	str	The parent intent ID.	required
title	str	Human-readable title for the child intent.	required
description	str	Detailed description of the goal.	''
constraints	Optional[dict[str, Any]]	Optional constraints dictionary (e.g. {"rules": [...]}).	None
initial_state	Optional[dict[str, Any]]	Optional initial state data.	None
depends_on	Optional[list[str]]	Optional list of intent IDs this depends on.	None

Returns:

Type	Description
Intent	The created child Intent object.

update_state

update_state(intent_id: str, version: int, state_patch: dict[str, Any]) -> Intent

Update intent state with optimistic concurrency control.

Parameters:

Name	Type	Description	Default
intent_id	str	The intent to update.	required
version	int	Expected current version (for conflict detection).	required
state_patch	dict[str, Any]	Partial state update to merge.	required

Returns:

Type	Description
Intent	The updated Intent object.

Raises:

Type	Description
ConflictError	If version doesn't match (another update occurred).

set_status

set_status(intent_id: str, version: int, status: IntentStatus, reason: Optional[str] = None) -> Intent

Change intent status.

Parameters:

Name	Type	Description	Default
intent_id	str	The intent to update.	required
version	int	Expected current version.	required
status	IntentStatus	New status to set.	required
reason	Optional[str]	Optional reason for the status change (required when governance policy sets require_status_reason=True).	None

Returns:

Type	Description
Intent	The updated Intent object.

log_event

log_event(intent_id: str, event_type: EventType, payload: Optional[dict[str, Any]] = None, trace_id: Optional[str] = None, parent_event_id: Optional[str] = None) -> IntentEvent

Append an event to the intent's audit log.

Parameters:

Name	Type	Description	Default
intent_id	str	The intent to log against.	required
event_type	EventType	Type of event.	required
payload	Optional[dict[str, Any]]	Event-specific data.	None
trace_id	Optional[str]	RFC-0020 correlation ID for distributed tracing.	None
parent_event_id	Optional[str]	RFC-0020 ID of the event that caused this one.	None

Returns:

Type	Description
IntentEvent	The created IntentEvent object.

get_events

get_events(intent_id: str, event_type: Optional[EventType] = None, since: Optional[datetime] = None, limit: int = 100) -> list[IntentEvent]

Retrieve events from the intent's audit log.

Parameters:

Name	Type	Description	Default
intent_id	str	The intent to query.	required
event_type	Optional[EventType]	Optional filter by event type.	None
since	Optional[datetime]	Optional filter for events after this time.	None
limit	int	Maximum number of events.	100

Returns:

Type	Description
list[IntentEvent]	List of IntentEvent objects.

acquire_lease

acquire_lease(intent_id: str, scope: str, duration_seconds: int = 300) -> IntentLease

Acquire a lease for exclusive access to a scope.

Parameters:

Name	Type	Description	Default
intent_id	str	The intent to lease within.	required
scope	str	The scope to acquire (e.g., "research", "synthesis").	required
duration_seconds	int	How long the lease should last.	300

Returns:

Type	Description
IntentLease	The acquired IntentLease object.

Raises:

Type	Description
LeaseConflictError	If scope is already leased by another agent.

release_lease

release_lease(intent_id: str, lease_id: str) -> None

Release a previously acquired lease.

Parameters:

Name	Type	Description	Default
intent_id	str	The intent containing the lease.	required
lease_id	str	The lease to release.	required

get_leases

get_leases(intent_id: str) -> list[IntentLease]

List all active leases for an intent.

Parameters:

Name	Type	Description	Default
intent_id	str	The intent to query.	required

Returns:

Type	Description
list[IntentLease]	List of IntentLease objects.

renew_lease

renew_lease(intent_id: str, lease_id: str, duration_seconds: int = 300) -> IntentLease

Renew an existing lease to extend its expiration.

Parameters:

Name	Type	Description	Default
intent_id	str	The intent the lease belongs to.	required
lease_id	str	The lease to renew.	required
duration_seconds	int	New duration from now.	300

Returns:

Type	Description
IntentLease	The renewed IntentLease object.

lease

lease(intent_id: str, scope: str, duration_seconds: int = 300) -> Generator[IntentLease, None, None]

Context manager for lease acquisition and release.

Example

with client.lease(intent_id, "analysis") as lease:
    # Perform exclusive work
    pass
# Lease automatically released

request_arbitration

request_arbitration(intent_id: str, reason: str, context: Optional[dict[str, Any]] = None) -> ArbitrationRequest

Request human arbitration for a conflict or decision.

Parameters:

Name	Type	Description	Default
intent_id	str	The intent requiring arbitration.	required
reason	str	Explanation of why arbitration is needed.	required
context	Optional[dict[str, Any]]	Additional context for the arbitrator.	None

Returns:

Type	Description
ArbitrationRequest	The created ArbitrationRequest object.

record_decision

record_decision(intent_id: str, decision_type: str, outcome: str, reasoning: str) -> Decision

Record a governance decision.

Parameters:

Name	Type	Description	Default
intent_id	str	The intent the decision applies to.	required
decision_type	str	Type of decision (e.g., "arbitration", "escalation").	required
outcome	str	The decision outcome.	required
reasoning	str	Explanation of the decision.	required

Returns:

Type	Description
Decision	The created Decision object.

get_decisions

get_decisions(intent_id: str) -> list[Decision]

Retrieve all decisions for an intent.

Parameters:

Name	Type	Description	Default
intent_id	str	The intent to query.	required

Returns:

Type	Description
list[Decision]	List of Decision objects.

assign_agent

assign_agent(intent_id: str, agent_id: Optional[str] = None) -> dict

Assign an agent to work on an intent.

Parameters:

Name	Type	Description	Default
intent_id	str	The intent to assign to.	required
agent_id	Optional[str]	Agent to assign (defaults to current agent).	None

Returns:

Type	Description
dict	Assignment confirmation.

unassign_agent

unassign_agent(intent_id: str, agent_id: Optional[str] = None) -> None

Remove an agent from an intent.

Parameters:

Name	Type	Description	Default
intent_id	str	The intent to unassign from.	required
agent_id	Optional[str]	Agent to remove (defaults to current agent).	None

create_portfolio

create_portfolio(name: str, description: Optional[str] = None, governance_policy: Optional[dict[str, Any]] = None, metadata: Optional[dict[str, Any]] = None) -> IntentPortfolio

Create a new intent portfolio for multi-intent coordination.

Parameters:

Name	Type	Description	Default
name	str	Portfolio name.	required
description	Optional[str]	Optional description.	None
governance_policy	Optional[dict[str, Any]]	Optional governance rules (e.g., require_all_completed).	None
metadata	Optional[dict[str, Any]]	Optional metadata.	None

Returns:

Type	Description
IntentPortfolio	The created portfolio.

get_portfolio

get_portfolio(portfolio_id: str) -> IntentPortfolio

Get a portfolio with its intents and aggregate status.

Parameters:

Name	Type	Description	Default
portfolio_id	str	The portfolio ID.	required

Returns:

Type	Description
IntentPortfolio	Portfolio with intents and aggregate status.

list_portfolios

list_portfolios(created_by: Optional[str] = None) -> list[IntentPortfolio]

List portfolios, optionally filtered by creator.

Parameters:

Name	Type	Description	Default
created_by	Optional[str]	Optional filter by creator.	None

Returns:

Type	Description
list[IntentPortfolio]	List of portfolios.

update_portfolio_status

update_portfolio_status(portfolio_id: str, status: PortfolioStatus) -> IntentPortfolio

Update portfolio status.

Parameters:

Name	Type	Description	Default
portfolio_id	str	The portfolio ID.	required
status	PortfolioStatus	New status (active, completed, abandoned).	required

Returns:

Type	Description
IntentPortfolio	Updated portfolio.

add_intent_to_portfolio

add_intent_to_portfolio(portfolio_id: str, intent_id: str, role: MembershipRole = MembershipRole.MEMBER, priority: int = 0) -> PortfolioMembership

Add an intent to a portfolio.

Parameters:

Name	Type	Description	Default
portfolio_id	str	The portfolio ID.	required
intent_id	str	The intent to add.	required
role	MembershipRole	Membership role (primary, member, dependency).	MEMBER
priority	int	Priority (higher = more important).	0

Returns:

Type	Description
PortfolioMembership	Membership record.

remove_intent_from_portfolio

remove_intent_from_portfolio(portfolio_id: str, intent_id: str) -> None

Remove an intent from a portfolio.

Parameters:

Name	Type	Description	Default
portfolio_id	str	The portfolio ID.	required
intent_id	str	The intent to remove.	required

get_portfolio_intents

get_portfolio_intents(portfolio_id: str) -> tuple[list[Intent], AggregateStatus]

Get all intents in a portfolio with aggregate status.

Parameters:

Name	Type	Description	Default
portfolio_id	str	The portfolio ID.	required

Returns:

Type	Description
tuple[list[Intent], AggregateStatus]	Tuple of (intents list, aggregate status).

subscribe

subscribe(intent_id: Optional[str] = None, portfolio_id: Optional[str] = None, event_types: Optional[list[str]] = None, webhook_url: Optional[str] = None, expires_at: Optional[datetime] = None) -> IntentSubscription

Subscribe to real-time notifications for an intent or portfolio.

Parameters:

Name	Type	Description	Default
intent_id	Optional[str]	Intent to subscribe to (optional).	None
portfolio_id	Optional[str]	Portfolio to subscribe to (optional).	None
event_types	Optional[list[str]]	Which events to receive (optional, all if not specified).	None
webhook_url	Optional[str]	URL to receive webhook notifications.	None
expires_at	Optional[datetime]	When subscription expires (optional).	None

Returns:

Type	Description
IntentSubscription	The created subscription.

log_llm_request_started

log_llm_request_started(intent_id: str, request_id: str, provider: str, model: str, messages_count: int, tools_available: Optional[list[str]] = None, stream: bool = False, temperature: Optional[float] = None, max_tokens: Optional[int] = None) -> IntentEvent

Log the start of an LLM API request.

Parameters:

Name	Type	Description	Default
intent_id	str	The intent this request is part of.	required
request_id	str	Unique identifier for this request.	required
provider	str	LLM provider (e.g., "openai", "anthropic").	required
model	str	Model being called.	required
messages_count	int	Number of messages in the request.	required
tools_available	Optional[list[str]]	List of tool names available to the model.	None
stream	bool	Whether this is a streaming request.	False
temperature	Optional[float]	Temperature setting.	None
max_tokens	Optional[int]	Max tokens setting.	None

Returns:

Type	Description
IntentEvent	The created event.

log_llm_request_completed

log_llm_request_completed(intent_id: str, request_id: str, provider: str, model: str, messages_count: int, response_content: Optional[str] = None, tool_calls: Optional[list[dict[str, Any]]] = None, finish_reason: Optional[str] = None, prompt_tokens: Optional[int] = None, completion_tokens: Optional[int] = None, total_tokens: Optional[int] = None, duration_ms: Optional[int] = None) -> IntentEvent

Log the successful completion of an LLM API request.

Parameters:

Name	Type	Description	Default
intent_id	str	The intent this request is part of.	required
request_id	str	Unique identifier for this request.	required
provider	str	LLM provider.	required
model	str	Model that was called.	required
messages_count	int	Number of messages in the request.	required
response_content	Optional[str]	Text content of the response.	None
tool_calls	Optional[list[dict[str, Any]]]	Any tool calls in the response.	None
finish_reason	Optional[str]	Why the model stopped generating.	None
prompt_tokens	Optional[int]	Tokens used for the prompt.	None
completion_tokens	Optional[int]	Tokens used for the completion.	None
total_tokens	Optional[int]	Total tokens used.	None
duration_ms	Optional[int]	Request duration in milliseconds.	None

Returns:

Type	Description
IntentEvent	The created event.

log_llm_request_failed

log_llm_request_failed(intent_id: str, request_id: str, provider: str, model: str, messages_count: int, error: str, duration_ms: Optional[int] = None) -> IntentEvent

Log a failed LLM API request.

Parameters:

Name	Type	Description	Default
intent_id	str	The intent this request is part of.	required
request_id	str	Unique identifier for this request.	required
provider	str	LLM provider.	required
model	str	Model that was called.	required
messages_count	int	Number of messages in the request.	required
error	str	Error message or description.	required
duration_ms	Optional[int]	Time until failure.	None

Returns:

Type	Description
IntentEvent	The created event.

log_tool_call_started

log_tool_call_started(intent_id: str, tool_name: str, tool_id: str, arguments: dict[str, Any], provider: Optional[str] = None, model: Optional[str] = None, parent_request_id: Optional[str] = None) -> IntentEvent

Log the start of a tool call initiated by an LLM.

Parameters:

Name	Type	Description	Default
intent_id	str	The intent this tool call is part of.	required
tool_name	str	Name of the tool being called.	required
tool_id	str	Unique identifier for this tool call.	required
arguments	dict[str, Any]	Arguments passed to the tool.	required
provider	Optional[str]	LLM provider (e.g., "openai", "anthropic").	None
model	Optional[str]	Model that initiated the call.	None
parent_request_id	Optional[str]	ID of the parent LLM request.	None

Returns:

Type	Description
IntentEvent	The created event.

log_tool_call_completed

log_tool_call_completed(intent_id: str, tool_name: str, tool_id: str, arguments: dict[str, Any], result: Any, duration_ms: Optional[int] = None, provider: Optional[str] = None, model: Optional[str] = None) -> IntentEvent

Log the successful completion of a tool call.

Parameters:

Name	Type	Description	Default
intent_id	str	The intent this tool call is part of.	required
tool_name	str	Name of the tool that was called.	required
tool_id	str	Unique identifier for this tool call.	required
arguments	dict[str, Any]	Arguments that were passed to the tool.	required
result	Any	Result returned by the tool.	required
duration_ms	Optional[int]	How long the tool call took.	None
provider	Optional[str]	LLM provider.	None
model	Optional[str]	Model that initiated the call.	None

Returns:

Type	Description
IntentEvent	The created event.

log_tool_call_failed

log_tool_call_failed(intent_id: str, tool_name: str, tool_id: str, arguments: dict[str, Any], error: str, duration_ms: Optional[int] = None, provider: Optional[str] = None, model: Optional[str] = None) -> IntentEvent

Log a failed tool call.

Parameters:

Name	Type	Description	Default
intent_id	str	The intent this tool call is part of.	required
tool_name	str	Name of the tool that failed.	required
tool_id	str	Unique identifier for this tool call.	required
arguments	dict[str, Any]	Arguments that were passed to the tool.	required
error	str	Error message or description.	required
duration_ms	Optional[int]	How long before the failure.	None
provider	Optional[str]	LLM provider.	None
model	Optional[str]	Model that initiated the call.	None

Returns:

Type	Description
IntentEvent	The created event.

invoke_tool

invoke_tool(tool_name: str, agent_id: str, parameters: Optional[dict[str, Any]] = None, **kwargs: Any) -> dict

Invoke a tool through the server's tool proxy (RFC-0014).

The server resolves the agent's grant, retrieves the credential, executes the tool, and records the invocation.

discover

discover() -> dict

Discover protocol capabilities via .well-known endpoint.

Returns:

Type	Description
dict	Protocol metadata including version, endpoints, and capabilities.

AsyncOpenIntentClient

The asynchronous client for async applications, with full RFC 0012-0017 support.

AsyncOpenIntentClient

AsyncOpenIntentClient(base_url: str, api_key: str, agent_id: str, timeout: float = 30.0)

Asynchronous client for the OpenIntent Coordination Protocol.

Example

async with AsyncOpenIntentClient(
    base_url="https://api.openintent.ai",
    api_key="your-api-key",
    agent_id="my-agent"
) as client:
    intent = await client.create_intent(
        title="Research task",
        description="Analyze data"
    )

create_intent async

create_intent(title: str, description: str, constraints: Optional[dict[str, Any]] = None, initial_state: Optional[dict[str, Any]] = None, governance_policy: Optional[dict[str, Any]] = None) -> Intent

Create a new intent.

get_intent async

get_intent(intent_id: str) -> Intent

Retrieve an intent by ID.

update_state async

update_state(intent_id: str, version: int, state_patch: dict[str, Any]) -> Intent

Update intent state with optimistic concurrency control.

log_event async

log_event(intent_id: str, event_type: EventType, payload: Optional[dict[str, Any]] = None, trace_id: Optional[str] = None, parent_event_id: Optional[str] = None) -> IntentEvent

Append an event to the intent's audit log.

Parameters:

Name	Type	Description	Default
trace_id	Optional[str]	RFC-0020 correlation ID for distributed tracing.	None
parent_event_id	Optional[str]	RFC-0020 ID of the event that caused this one.	None

subscribe async

subscribe(webhook_url: str, event_types: Optional[list[str]] = None, intent_id: Optional[str] = None, portfolio_id: Optional[str] = None, expires_in_hours: int = 24) -> IntentSubscription

Subscribe to events for an intent or portfolio.

create_task async

create_task(intent_id: str, name: str, **kwargs: Any) -> Task

Create a task for an intent (RFC-0012).

get_task async

get_task(task_id: str) -> Task

Get a task by ID (RFC-0012).

list_tasks async

list_tasks(intent_id: str, status: Optional[TaskStatus] = None, limit: int = 100, offset: int = 0) -> list[Task]

List tasks for an intent (RFC-0012).

update_task async

update_task(task_id: str, version: int, status: Optional[str] = None, **kwargs: Any) -> Task

Update task status with optimistic concurrency (RFC-0012).

create_plan async

create_plan(intent_id: str, tasks: Optional[list[str]] = None, **kwargs: Any) -> Plan

Create a plan for an intent (RFC-0012).

get_plan async

get_plan(plan_id: str) -> Plan

Get a plan by ID (RFC-0012).

list_plans async

list_plans(intent_id: str) -> list[Plan]

List plans for an intent (RFC-0012).

update_plan async

update_plan(plan_id: str, version: int, **kwargs: Any) -> Plan

Update a plan with optimistic concurrency (RFC-0012).

create_coordinator_lease async

create_coordinator_lease(agent_id: str, intent_id: Optional[str] = None, **kwargs: Any) -> CoordinatorLease

Create a coordinator lease (RFC-0013).

get_coordinator_lease async

get_coordinator_lease(lease_id: str) -> CoordinatorLease

Get a coordinator lease (RFC-0013).

list_coordinator_leases async

list_coordinator_leases(intent_id: Optional[str] = None) -> list[CoordinatorLease]

List coordinator leases (RFC-0013).

coordinator_heartbeat async

coordinator_heartbeat(lease_id: str) -> dict

Send coordinator heartbeat (RFC-0013).

create_decision_record async

create_decision_record(coordinator_id: str, intent_id: str, decision_type: DecisionType, summary: str, rationale: str, **kwargs: Any) -> dict

Record a coordination decision (RFC-0013).

list_decision_records async

list_decision_records(intent_id: str, limit: int = 50) -> list[dict]

List decision records for an intent (RFC-0013).

create_vault async

create_vault(owner_id: str, name: str) -> dict

Create a credential vault (RFC-0014).

get_vault async

get_vault(vault_id: str) -> dict

Get a credential vault (RFC-0014).

create_credential async

create_credential(vault_id: str, service: str, label: str, auth_type: str = 'api_key', **kwargs: Any) -> dict

Create a credential in a vault (RFC-0014).

get_credential async

get_credential(credential_id: str) -> dict

Get a credential (RFC-0014).

create_tool_grant async

create_tool_grant(credential_id: str, agent_id: str, granted_by: str, scopes: Optional[list[str]] = None, **kwargs: Any) -> ToolGrant

Create a tool grant (RFC-0014).

get_tool_grant async

get_tool_grant(grant_id: str) -> ToolGrant

Get a tool grant (RFC-0014).

list_agent_grants async

list_agent_grants(agent_id: str) -> list[ToolGrant]

List grants for an agent (RFC-0014).

revoke_grant async

revoke_grant(grant_id: str) -> dict

Revoke a tool grant (RFC-0014).

record_invocation async

record_invocation(grant_id: str, service: str, tool: str, agent_id: str, **kwargs: Any) -> dict

Record a tool invocation (RFC-0014).

list_invocations async

list_invocations(grant_id: str, limit: int = 50) -> list[dict]

List invocations for a grant (RFC-0014).

create_memory async

create_memory(agent_id: str, namespace: str, key: str, value: dict, memory_type: str = 'working', **kwargs: Any) -> MemoryEntry

Create a memory entry (RFC-0015).

get_memory async

get_memory(entry_id: str) -> MemoryEntry

Get a memory entry (RFC-0015).

list_memory async

list_memory(agent_id: str, namespace: Optional[str] = None, memory_type: Optional[str] = None, tags: Optional[list[str]] = None, limit: int = 100, offset: int = 0) -> list[MemoryEntry]

List memory entries for an agent (RFC-0015).

update_memory async

update_memory(entry_id: str, version: int, **kwargs: Any) -> MemoryEntry

Update a memory entry with optimistic concurrency (RFC-0015).

delete_memory async

delete_memory(entry_id: str) -> None

Delete a memory entry (RFC-0015).

register_agent async

register_agent(agent_id: str, capabilities: Optional[list[str]] = None, **kwargs: Any) -> AgentRecord

Register an agent (RFC-0016).

get_agent_record async

get_agent_record(agent_id: str) -> AgentRecord

Get agent record (RFC-0016).

list_agents async

list_agents(status: Optional[str] = None, role_id: Optional[str] = None) -> list[AgentRecord]

List registered agents (RFC-0016).

agent_heartbeat async

agent_heartbeat(agent_id: str, current_load: int = 0, tasks_in_progress: Optional[list[str]] = None) -> dict

Send agent heartbeat (RFC-0016).

update_agent_status async

update_agent_status(agent_id: str, status: str) -> dict

Update agent status (RFC-0016).

create_trigger async

create_trigger(name: str, trigger_type: str, **kwargs: Any) -> Trigger

Create a trigger (RFC-0017).

get_trigger async

get_trigger(trigger_id: str) -> Trigger

Get a trigger (RFC-0017).

list_triggers async

list_triggers(namespace: Optional[str] = None, trigger_type: Optional[str] = None) -> list[Trigger]

List triggers (RFC-0017).

update_trigger async

update_trigger(trigger_id: str, version: int, **kwargs: Any) -> Trigger

Update a trigger with optimistic concurrency (RFC-0017).

fire_trigger async

fire_trigger(trigger_id: str) -> dict

Fire a trigger manually (RFC-0017).

delete_trigger async

delete_trigger(trigger_id: str) -> None

Delete a trigger (RFC-0017).

invoke_tool async

invoke_tool(tool_name: str, agent_id: str, parameters: Optional[dict[str, Any]] = None, **kwargs: Any) -> dict

Invoke a tool through the server's tool proxy (RFC-0014).

The server resolves the agent's grant, retrieves the credential, executes the tool, and records the invocation.

close async

close() -> None

Close the HTTP client connection.