Python SDK Reference

Official Python client for GL AIP (GDP Labs AI Agents Package). Typed, session-aware, and aligned with the live FastAPI backend.

This document is the authoritative guide to the glaip-sdk package. Every method, payload, and return type documented here is derived from the code in python/glaip-sdk/glaip_sdk and mirrors the behaviour used by the CLI and the end-to-end test suite.

The content is organised by resource category so you can quickly locate the operation you need—agents, tools, MCP connections, language models, and utility helpers.

Quickstart (SDK) — pip install glaip-sdk → export AIP_API_URL & AIP_API_KEY → with Client() as client: → create and run an agent.

---

Installation & Client Basics

pip install glaip-sdk

from glaip_sdk import Client

# Read configuration from environment variables (AIP_API_URL, AIP_API_KEY)
client = Client()

# Explicit configuration
client = Client(
    api_url="https://aip.example.com",
    api_key="sk-your-api-key",
    timeout=60.0,
)

Environment variables: The client reads AIP_API_URL and AIP_API_KEY by default when arguments are omitted.

Session lifecycle

Action	Behaviour
with Client() as client:	Opens and automatically closes the underlying httpx.Client.
client.timeout = 90	Rebuilds the shared HTTP client and propagates the session to agents, tools, and mcps.
client.close()	Manual teardown when not using a context manager.

with Client(timeout=45) as client:
    client.timeout = 90  # propagates to sub-clients
    print(client.agents.list_agents())

---

Client API Surface

The Client class exposes convenience methods that delegate to the specialised sub-clients under the hood. This section documents every public method grouped by resource.

Each signature is shown exactly as implemented. Optional parameters list their Python default values.

Agents

client.create_agent(name=None, instruction=None, model=None, tools=None, agents=None, timeout=None, *, file=None, **kwargs) -> Agent

Create an agent with the supplied configuration. When file is provided, the SDK loads the agent definition from a JSON or YAML document, merges any keyword overrides you pass, and submits the combined payload.

Parameter	Type	Required	Default	Example	Description
name	str	Yes (unless file supplied)	—	"pipeline-runner"	Human-readable agent name (must be unique per account). Loaded from file when omitted.
instruction	str	Yes (unless file supplied)	—	"Coordinate ETL tasks..."	System prompt / operating instructions. Pulled from file when omitted.
model	str	No	DEFAULT_MODEL (dev fallback only)	"gpt-5-nano"	Language model name when not using language_model_id. Leave unset to rely on your workspace default or pass language_model_id for catalog models.
tools	list[str | Tool] | None	No	None	["tool-uuid"]	Tool IDs or Tool objects to attach. IDs are extracted automatically.
agents	list[str | Agent] | None	No	None	["delegate-id"]	Sub-agent IDs or Agent objects to delegate work to.
timeout	int	No	DEFAULT_AGENT_RUN_TIMEOUT	90	Execution timeout (seconds). Stored as a string for backend compatibility. Overrides file contents when provided.
file	str | Path	No	None	"configs/agent.yaml"	Load base configuration from JSON/YAML. Overrides merge with file data (arrays like tools, agents, mcps are appended, not replaced).
**kwargs	—	No	—	language_model_id="models/vertex"	Forwarded to the backend payload (e.g. language_model_id, agent_config, metadata, mcps).

When file is used you may specify tools/agents/MCPs by name instead of ID; the SDK resolves them against the current workspace before submitting the payload (mirroring the CLI’s --import behaviour).

Returns the created Agent with all persisted fields populated. Raises ValueError if neither runtime arguments nor the file payload provide name and instruction.

Complete field specifications: See Agents Schema Reference for all available fields, validation rules, and constraints. Additional fields can be passed via **kwargs.

agent = client.create_agent(
    name="pipeline-runner",
    instruction="Coordinate ETL tasks and report results",
    tools=["tool-uuid"],
    metadata={"team": "ml"},
    agent_config={"memory": "mem0", "tool_output_sharing": True},
)

agent = client.create_agent(
    file="configs/sentiment.yaml",
    metadata={"environment": "prod"},
    tools=["tool-uuid-extra"],
)

client.list_agents(agent_type=None, framework=None, name=None, version=None, sync_langflow_agents=False, *, limit=None, page=None, include_deleted=None, created_at_start=None, created_at_end=None, updated_at_start=None, updated_at_end=None, metadata=None, query=None) -> AgentListResult

Fetch agents with optional filtering and pagination metadata. The returned AgentListResult is iterable like a list of Agent models and also exposes total, page, limit, has_next, and has_prev.

Filter	Type	Required	Default	Example	Description
agent_type	str	No	None	"langflow"	Filter by config, code, a2a, or langflow.
framework	str	No	None	"langgraph"	Filter by orchestration framework.
name	str	No	None	"pipeline"	Case-insensitive substring match.
version	str	No	None	"1.2.0"	Exact version.
sync_langflow_agents	bool	No	False	True	When True, performs a LangFlow sync before listing.
limit	int	No	None	20	Page size (1-100).
page	int	No	None	2	1-based page number.
include_deleted	bool	No	None	True	Include soft-deleted agents.
created_at_start / created_at_end	str	No	None	"2024-01-01T00:00:00Z"	Filter by creation timestamp range (ISO 8601).
updated_at_start / updated_at_end	str	No	None	"2024-02-01T00:00:00Z"	Filter by last update timestamp range (ISO 8601).
metadata	dict[str, str]	No	None	{"environment": "prod"}	Apply metadata filters (metadata.environment=prod).
query	AgentListParams	No	None	—	Supply a pre-built parameter object to avoid passing individual filters.

Returns an empty result set if no agents match (len(result) == 0).

client.get_agent_by_id(agent_id) -> Agent

Retrieve an agent by UUID. Raises NotFoundError if the agent does not exist or belongs to another account.

client.get_agent(agent_id) is an alias.

client.find_agents(name=None) -> list[Agent]

List agents and apply a case-insensitive name filter client-side. Returns an empty list if nothing matches.

client.update_agent(agent_id, name=None, instruction=None, model=None, *, file=None, **kwargs) -> Agent

Replace an agent's configuration.

• Unspecified parameters retain their stored values.

• Pass language_model_id via **kwargs to reference the shared model catalog.

• Provide tools=[] / agents=[] to clear associations.

• Supply file to load a JSON/YAML definition and merge overrides, matching the CLI's --import behaviour.

When updating from a file, the SDK also resolves tool/agent/MCP names to the corresponding IDs in the current workspace so you can reuse CLI exports without manual edits.

Complete field specifications: See Agents Schema Reference for all available fields, validation rules, and constraints. Additional fields can be passed via **kwargs.

updated = client.update_agent(
    agent.id,
    instruction="Be precise and provide references",
    timeout=900,
    metadata={"environment": "prod"},
)

client.delete_agent(agent_id) -> None

Soft-delete the agent.

client.run_agent(agent_id, message, files=None, tty=False, **kwargs) -> str

Execute an agent and stream results. Returns the final assistant response as a string.

Parameter	Type	Description
message	str	Prompt for the agent.
files	list[str | BinaryIO] | None	Optional file attachments. Provide file paths (strings).
tty	bool	When True, enables the Rich TTY renderer (auto by default on the CLI).
**kwargs	–	Forwarded to the backend payload (e.g. chat_history, pii_mapping, runtime_config, timeout, renderer).

The SDK automatically handles SSE streaming, rich rendering, and error handling. Mirrors POST /agents/{agent_id}/run.

response = client.run_agent(
    agent.id,
    "Summarise ticket INC-42",
    chat_history=[{"role": "user", "content": "Please continue"}]
)

client.sync_langflow_agents(base_url=None, api_key=None) -> dict

Trigger LangFlow synchronization. Returns the backend JSON response containing counts of created/updated flows. base_url and api_key override the environment variables LANGFLOW_BASE_URL / LANGFLOW_API_KEY when set. Pairs with POST /agents/langflow/sync.

Async streaming – client.agents.arun_agent(agent_id, message, files=None, timeout=None, **kwargs)

Asynchronous generator that yields parsed SSE JSON fragments. Useful for server applications or custom renderers.

async for event in client.agents.arun_agent(agent.id, "Status update"):
    print(event)

Example SSE payload emitted by the backend:

event: token
data: {"text": "Working on it..."}

event: completed
data: {"output": "Final answer here"}

Agent model helpers

Agents returned by the SDK are Pydantic models that keep a reference to the originating client when fetched through the SDK.

Method	Description
agent.run(message, **kwargs) -> str	Delegates to client.run_agent, injecting the agent name automatically.
agent.update(**kwargs) -> Agent	Calls client.update_agent and refreshes model fields in-place.
agent.delete() -> None	Calls client.delete_agent.

agent = client.get_agent_by_id(agent_id)
result = agent.run("Summarise ticket INC-42", timeout=120)
print(result)

---

Tools

client.create_tool(file_path, name=None, description=None, framework="langchain", **kwargs) -> Tool

Upload a Python plugin packaged as a .py file.

Parameter	Type	Description
file_path	str	Path to the plugin file. Must exist locally.
name	str | None	Optional override; default is derived from the plugin’s name attribute.
description	str | None	Stored description. Auto-generated if omitted.
framework	str	Tool framework identifier (langchain default).
**kwargs	–	Additional metadata (e.g. tags, tool_type).

Returns the created Tool instance. Temporary files created during upload are cleaned up automatically.

Complete field specifications: See Tools Schema Reference for all available fields, validation rules, and constraints.

tool = client.create_tool(
    file_path="calculator_tool.py",
    tags=["math", "utility"],
)

client.list_tools(tool_type=None) -> list[Tool]

List tool metadata. Optional tool_type filters by backend type (custom, native).

client.get_tool_by_id(tool_id) -> Tool

Fetch a tool by UUID. Raises NotFoundError if missing.

client.get_tool(tool_id) is an alias.

client.find_tools(name) -> list[Tool]

Return tools whose names match the supplied string (case-insensitive, client-side filter).

client.update_tool(tool_id, **kwargs) -> Tool

Update tool metadata (e.g. name, description, tags). For custom tools that need a code refresh, use client.update_tool_via_file.

client.update_tool_via_file(tool_id, file_path, **kwargs) -> Tool

Upload a new plugin file for an existing custom tool. Any kwargs provided are forwarded as form fields (description, tags, etc.).

client.get_tool_script(tool_id) -> str

Return the stored plugin source. Useful for audits or version control.

client.delete_tool(tool_id) -> None

Soft-delete the tool.

Tool model helpers

Method	Description
tool.get_script() -> str	Returns cached tool_script or a placeholder when not available.
tool.update(**kwargs) -> Tool	Delegates to the appropriate SDK update method (metadata vs file upload).
tool.delete() -> None	Delegates to client.delete_tool.

---

Model Context Protocol (MCP)

client.create_mcp(**kwargs) -> MCP

Create an MCP configuration. Accepts the same payload shape as the REST API (/mcps POST) including name, description, transport, config, and authentication dictionaries.

Complete field specifications: See MCP Schema Reference for all available fields, validation rules, and constraints.

mcp = client.create_mcp(
    name="vector-db",
    transport="http",
    config={"url": "https://mcp.example.com"},
    authentication={"type": "api-key", "key": "X-API-Key", "value": "secret"},
)

client.list_mcps() -> list[MCP]

Return all stored MCP configurations for the account.

client.get_mcp_by_id(mcp_id) -> MCP

Retrieve an MCP by UUID. client.get_mcp(mcp_id) is an alias.

client.find_mcps(name) -> list[MCP]

Client-side name filtering across MCP configurations.

client.update_mcp(mcp_id, **kwargs) -> MCP

Update an MCP configuration. The SDK chooses between PUT or PATCH depending on whether you provide a full payload (name, config, transport).

Complete field specifications: See MCP Schema Reference for all available fields, validation rules, and constraints.

client.delete_mcp(mcp_id) -> None

Soft-delete an MCP.

client.test_mcp_connection(config: dict) -> dict

Call /mcps/connect to validate credentials or connectivity without persisting the configuration.

client.test_mcp_connection_from_config(config) is an alias.

client.get_mcp_tools_from_config(config: dict) -> list[dict]

Call /mcps/connect/tools to fetch tool metadata from an MCP without storing it. Useful for validating available actions before creation.

MCP model helpers

Method	Description
mcp.update(**kwargs) -> MCP	Delegates to client.update_mcp.
mcp.delete() -> None	Delegates to client.delete_mcp.

---

Schedules

client.schedules.create(*, agent_id, input, schedule) -> Schedule

Create a schedule for recurring agent execution.

Parameter	Type	Required	Default	Example	Description
agent_id	str	Yes	—	"agent-123"	Agent ID to schedule runs for.
input	str	Yes	—	"Generate daily report"	Input text for each scheduled execution.
schedule	ScheduleConfig | dict | str	Yes	—	"0 9 * * 0-4"	Cron config as object, dict, or string.

Cron strings use five fields: minute hour day_of_month month day_of_week. Fields accept *, ranges (e.g., 2-4), lists (0,6), and steps (*/N). Day-of-week uses APScheduler numbering (0 = Monday, 6 = Sunday). All schedules run in Asia/Jakarta (WIB).

Returns the created Schedule instance.

from glaip_sdk.models.schedule import ScheduleConfig

schedule = client.schedules.create(
    agent_id="agent-123",
    input="Generate daily summary",
    schedule=ScheduleConfig(
        minute="0",
        hour="9",
        day_of_week="0-4",  # Weekdays (Mon-Fri)
    ),
)

# Or use a cron string directly
schedule = client.schedules.create(
    agent_id="agent-123",
    input="Weekly update",
    schedule="0 10 * * 0",  # Every Monday at 10am
)

client.schedules.list(*, limit=None, page=None, agent_id=None) -> ScheduleListResult

List schedules with optional filtering and pagination.

Parameter	Type	Required	Default	Description
limit	int	No	None	Page size (1-100).
page	int	No	None	Page number for pagination.
agent_id	str	No	None	Filter schedules by agent ID.

Returns ScheduleListResult with items, total, page, limit, has_next, has_prev.

# List all schedules
schedules = client.schedules.list()

# Filter by agent
agent_schedules = client.schedules.list(agent_id="agent-123")

# Paginate
page1 = client.schedules.list(limit=10, page=1)

client.schedules.get(schedule_id) -> Schedule

Retrieve a schedule by UUID.

Raises NotFoundError if the schedule does not exist.

from glaip_sdk.exceptions import NotFoundError

try:
    schedule = client.schedules.get("schedule-abc")
    print(schedule.next_run_time)
except NotFoundError:
    print("Schedule not found")

client.schedules.update(schedule_id, *, input=None, schedule=None) -> Schedule

Update an existing schedule. Unspecified parameters retain their values.

Update Behavior (Explicit, Not Merge):

• Omit schedule to keep the existing timing unchanged

• Providing a partial schedule dict (e.g., {"hour": "10"}) will fill missing fields with "*"

• This is intentional for predictability: what you provide is what you get, plus wildcard defaults

• To preserve existing values, fetch the current schedule first and modify only what you need

Parameter	Type	Required	Default	Description
schedule_id	str	Yes	—	Schedule ID to update.
input	str	No	None	New input text for scheduled execution.
schedule	ScheduleConfig | dict | str	No	None	New schedule configuration.

updated = client.schedules.update(
    "schedule-abc",
    input="Updated daily report",
    schedule="0 8 * * *",  # Change to 8am
)

client.schedules.delete(schedule_id) -> None

Delete a schedule by ID.

client.schedules.delete("schedule-abc")

client.schedules.list_runs(agent_id, *, schedule_id=None, status=None, limit=None, page=None) -> ScheduleRunListResult

List execution runs for an agent, optionally filtered by schedule and status. Only returns schedule runs (equivalent to run_type=schedule).

Parameter	Type	Required	Default	Description
agent_id	str	Yes	—	Agent ID to list runs for.
schedule_id	str	No	None	Filter by specific schedule ID.
status	RunStatus	No	None	Filter by run status. Valid values: started, success, failed, cancelled, aborted, unavailable (lowercase).
limit	int	No	None	Page size (1-100).
page	int	No	None	Page number.

# List all scheduled runs for an agent
runs = client.schedules.list_runs("agent-123")

# Filter by schedule and status
successful_runs = client.schedules.list_runs(
    "agent-123",
    schedule_id="schedule-abc",
    status="success",
)

Schedule model helpers

Method	Description
schedule.update(**kwargs) -> Schedule	Update schedule config or input.
schedule.delete() -> None	Delete the schedule.
schedule.list_runs(**params) -> ScheduleRunListResult	List runs for this schedule.
run.get_result() -> ScheduleRunResult	Fetch full output for a run.
schedule_run.duration -> str | None	Formatted duration (HH:MM:SS).

Schedule run fields

Schedule run list items (ScheduleRun) expose run metadata:

Field	Description
id	Run ID.
agent_id	Agent ID associated with the run.
schedule_id	Schedule ID for scheduled runs.
run_type	Run type, usually schedule.
status	Run status (lowercase): started, success, failed, cancelled, aborted, unavailable.
started_at	Execution start timestamp.
completed_at	Execution end timestamp.
input	Input used for the run, when provided by the backend.
config	Schedule config used for the run, when provided by the backend.

Schedule run results

run.get_result() returns a ScheduleRunResult with the stored output payload. The output field is a list of backend-defined event dictionaries and is not guaranteed to match a fixed schema.

Agent schedule facade

Access agent.schedule for scoped operations:

agent = client.get_agent_by_id("agent-123")

# Create schedule via agent
schedule = agent.schedule.create(
    input="Daily task",
    schedule="0 9 * * 0-4",
)

# List this agent's schedules
for s in agent.schedule.list():
    print(s.id)

# List runs for a schedule
runs = agent.schedule.list_runs(schedule.id)

---

Language Models & Utilities

client.list_language_models() -> list[dict]

Returns the language model catalogue visible to the current API key. Each entry contains provider metadata, model identifiers, and optional base URLs.

client.ping() -> bool

Calls /health-check. Returns True when the API responds successfully, otherwise False.

client.timeout

Property exposing the current timeout (seconds). Assigning a new value rebuilds the shared httpx.Client instance.

---

Error Handling

All network operations raise typed exceptions from glaip_sdk.exceptions when the backend responds with an error status.

Exception	Trigger
AuthenticationError	Missing/invalid X-API-Key (401).
ForbiddenError	Attempt to access a master-key-only endpoint with an account key (403).
NotFoundError	Resource does not exist or is soft-deleted (404).
ValidationError	Payload failed validation (400/422).
ConflictError	Duplicate names or incompatible state (409).
RateLimitError	Too many requests (429).
TimeoutError / AgentTimeoutError	Request or streaming timeout.
ServerError	Backend 5xx responses.

Example pattern:

from glaip_sdk import Client
from glaip_sdk.exceptions import AuthenticationError, ValidationError

client = Client()

try:
    agent = client.create_agent(name="demo", instruction="Be helpful")
except AuthenticationError:
    print("Invalid API key")
except ValidationError as exc:
    print("Payload rejected", exc.payload)

Troubleshooting & FAQ

• 401 Invalid API key — Refresh credentials from the AIP console and confirm AIP_API_KEY is set.

• 404 Not found — Resources are soft-deleted; call client.list_* to confirm IDs before retrying.

• 409 Conflict — Resolve duplicate names or finish outstanding runs before retrying the write.

• 429 Rate limited — Catch RateLimitError and back off using the Retry-After header from the REST response.

• Streaming stalls — Increase timeout; SSE connections idle for 5 minutes are closed server-side.

---

Related Documentation

• CLI Commands Reference

• REST API Reference

• Built-in examples in python/glaip-sdk/examples/

Keep this page in sync whenever you add new client methods, adjust signatures, or expand the backend payloads consumed by the SDK.