Compare commits

...

2 Commits

Author SHA1 Message Date
Reinier van der Leer
a301ae4879 Merge branch 'dev' into pwuts/open-2923-v2-external-api 2026-01-16 14:48:02 +01:00
Reinier van der Leer
87a63ecdba first draft (missing migrations) 2026-01-05 15:30:01 +01:00
21 changed files with 3479 additions and 83 deletions

View File

@@ -1,21 +1,57 @@
from fastapi import FastAPI
"""
External API Application
This module defines the main FastAPI application for the external API,
which mounts the v1 and v2 sub-applications.
"""
from fastapi import FastAPI
from fastapi.responses import RedirectResponse
from backend.api.middleware.security import SecurityHeadersMiddleware
from backend.monitoring.instrumentation import instrument_fastapi
from .v1.routes import v1_router
from .v1.app import v1_app
from .v2.app import v2_app
DESCRIPTION = """
The external API provides programmatic access to the AutoGPT Platform for building
integrations, automations, and custom applications.
### API Versions
| Version | End of Life | Path | Documentation |
|---------------------|-------------|------------------------|---------------|
| **v2** | | `/external-api/v2/...` | [v2 docs](v2/docs) |
| **v1** (deprecated) | 2025-05-01 | `/external-api/v1/...` | [v1 docs](v1/docs) |
**Recommendation**: New integrations should use v2.
For authentication details and usage examples, see the
[API Integration Guide](https://docs.agpt.co/platform/integrating/api-guide/).
"""
external_api = FastAPI(
title="AutoGPT External API",
description="External API for AutoGPT integrations",
title="AutoGPT Platform API",
summary="External API for AutoGPT Platform integrations",
description=DESCRIPTION,
version="2.0.0",
docs_url="/docs",
version="1.0",
redoc_url="/redoc",
)
external_api.add_middleware(SecurityHeadersMiddleware)
external_api.include_router(v1_router, prefix="/v1")
# Add Prometheus instrumentation
@external_api.get("/", include_in_schema=False)
async def root_redirect() -> RedirectResponse:
"""Redirect root to API documentation."""
return RedirectResponse(url="/docs")
# Mount versioned sub-applications
# Each sub-app has its own /docs page at /v1/docs and /v2/docs
external_api.mount("/v1", v1_app)
external_api.mount("/v2", v2_app)
# Add Prometheus instrumentation to the main app
instrument_fastapi(
external_api,
service_name="external-api",

View File

@@ -0,0 +1,39 @@
"""
V1 External API Application
This module defines the FastAPI application for the v1 external API.
"""
from fastapi import FastAPI
from backend.api.middleware.security import SecurityHeadersMiddleware
from .routes import v1_router
DESCRIPTION = """
The v1 API provides access to core AutoGPT functionality for external integrations.
For authentication details and usage examples, see the
[API Integration Guide](https://docs.agpt.co/platform/integrating/api-guide/).
"""
v1_app = FastAPI(
title="AutoGPT Platform API",
summary="External API for AutoGPT Platform integrations (v1)",
description=DESCRIPTION,
version="1.0.0",
docs_url="/docs",
redoc_url="/redoc",
openapi_url="/openapi.json",
openapi_tags=[
{"name": "user", "description": "User information"},
{"name": "blocks", "description": "Block operations"},
{"name": "graphs", "description": "Graph execution"},
{"name": "store", "description": "Marketplace agents and creators"},
{"name": "integrations", "description": "OAuth credential management"},
{"name": "tools", "description": "AI assistant tools"},
],
)
v1_app.add_middleware(SecurityHeadersMiddleware)
v1_app.include_router(v1_router)

View File

@@ -0,0 +1,9 @@
"""
V2 External API
This module provides the v2 external API for programmatic access to the AutoGPT Platform.
"""
from .routes import v2_router
__all__ = ["v2_router"]

View File

@@ -0,0 +1,82 @@
"""
V2 External API Application
This module defines the FastAPI application for the v2 external API.
"""
from fastapi import FastAPI
from backend.api.middleware.security import SecurityHeadersMiddleware
from .routes import v2_router
DESCRIPTION = """
The v2 API provides comprehensive access to the AutoGPT Platform for building
integrations, automations, and custom applications.
### Key Improvements over v1
- **Consistent naming**: Uses `graph_id`/`graph_version` consistently
- **Better pagination**: All list endpoints support pagination
- **Comprehensive coverage**: Access to library, runs, schedules, credits, and more
- **Human-in-the-loop**: Review and approve agent decisions via the API
For authentication details and usage examples, see the
[API Integration Guide](https://docs.agpt.co/platform/integrating/api-guide/).
### Pagination
List endpoints return paginated responses. Use `page` and `page_size` query
parameters to navigate results. Maximum page size is 100 items.
"""
v2_app = FastAPI(
title="AutoGPT Platform External API",
summary="External API for AutoGPT Platform integrations (v2)",
description=DESCRIPTION,
version="2.0.0",
docs_url="/docs",
redoc_url="/redoc",
openapi_url="/openapi.json",
openapi_tags=[
{
"name": "graphs",
"description": "Create, update, and manage agent graphs",
},
{
"name": "schedules",
"description": "Manage scheduled graph executions",
},
{
"name": "blocks",
"description": "Discover available building blocks",
},
{
"name": "marketplace",
"description": "Browse agents and creators, manage submissions",
},
{
"name": "library",
"description": "Access your agent library and execute agents",
},
{
"name": "runs",
"description": "Monitor execution runs and human-in-the-loop reviews",
},
{
"name": "credits",
"description": "Check balance and view transaction history",
},
{
"name": "integrations",
"description": "Manage OAuth credentials for external services",
},
{
"name": "files",
"description": "Upload files for agent input",
},
],
)
v2_app.add_middleware(SecurityHeadersMiddleware)
v2_app.include_router(v2_router)

View File

@@ -0,0 +1,140 @@
"""
V2 External API - Blocks Endpoints
Provides read-only access to available building blocks.
"""
import logging
from typing import Any
from fastapi import APIRouter, Response, Security
from fastapi.concurrency import run_in_threadpool
from prisma.enums import APIKeyPermission
from pydantic import BaseModel, Field
from backend.api.external.middleware import require_permission
from backend.data.auth.base import APIAuthorizationInfo
from backend.data.block import get_blocks
from backend.util.cache import cached
from backend.util.json import dumps
logger = logging.getLogger(__name__)
blocks_router = APIRouter()
# ============================================================================
# Models
# ============================================================================
class BlockCost(BaseModel):
"""Cost information for a block."""
cost_type: str = Field(description="Type of cost (e.g., 'per_call', 'per_token')")
cost_filter: dict[str, Any] = Field(
default_factory=dict, description="Conditions for this cost"
)
cost_amount: int = Field(description="Cost amount in credits")
class Block(BaseModel):
"""A building block that can be used in graphs."""
id: str
name: str
description: str
categories: list[str] = Field(default_factory=list)
input_schema: dict[str, Any]
output_schema: dict[str, Any]
costs: list[BlockCost] = Field(default_factory=list)
disabled: bool = Field(default=False)
class BlocksListResponse(BaseModel):
"""Response for listing blocks."""
blocks: list[Block]
total_count: int
# ============================================================================
# Internal Functions
# ============================================================================
def _compute_blocks_sync() -> str:
"""
Synchronous function to compute blocks data.
This does the heavy lifting: instantiate 226+ blocks, compute costs, serialize.
"""
from backend.data.credit import get_block_cost
block_classes = get_blocks()
result = []
for block_class in block_classes.values():
block_instance = block_class()
if not block_instance.disabled:
costs = get_block_cost(block_instance)
# Convert BlockCost BaseModel objects to dictionaries
costs_dict = [
cost.model_dump() if isinstance(cost, BaseModel) else cost
for cost in costs
]
result.append({**block_instance.to_dict(), "costs": costs_dict})
return dumps(result)
@cached(ttl_seconds=3600)
async def _get_cached_blocks() -> str:
"""
Async cached function with thundering herd protection.
On cache miss: runs heavy work in thread pool
On cache hit: returns cached string immediately
"""
return await run_in_threadpool(_compute_blocks_sync)
# ============================================================================
# Endpoints
# ============================================================================
@blocks_router.get(
path="",
summary="List available blocks",
responses={
200: {
"description": "List of available building blocks",
"content": {
"application/json": {
"schema": {
"items": {"additionalProperties": True, "type": "object"},
"type": "array",
}
}
},
}
},
)
async def list_blocks(
auth: APIAuthorizationInfo = Security(
require_permission(APIKeyPermission.READ_BLOCK)
),
) -> Response:
"""
List all available building blocks that can be used in graphs.
Each block represents a specific capability (e.g., HTTP request, text processing,
AI completion, etc.) that can be connected in a graph to create an agent.
The response includes input/output schemas for each block, as well as
cost information for blocks that consume credits.
"""
content = await _get_cached_blocks()
return Response(
content=content,
media_type="application/json",
)

View File

@@ -0,0 +1,36 @@
"""
Common utilities for V2 External API
"""
from typing import TypeVar
from pydantic import BaseModel, Field
# Constants for pagination
MAX_PAGE_SIZE = 100
DEFAULT_PAGE_SIZE = 20
class PaginationParams(BaseModel):
"""Common pagination parameters."""
page: int = Field(default=1, ge=1, description="Page number (1-indexed)")
page_size: int = Field(
default=DEFAULT_PAGE_SIZE,
ge=1,
le=MAX_PAGE_SIZE,
description=f"Number of items per page (max {MAX_PAGE_SIZE})",
)
T = TypeVar("T")
class PaginatedResponse(BaseModel):
"""Generic paginated response wrapper."""
items: list
total_count: int = Field(description="Total number of items across all pages")
page: int = Field(description="Current page number (1-indexed)")
page_size: int = Field(description="Number of items per page")
total_pages: int = Field(description="Total number of pages")

View File

@@ -0,0 +1,141 @@
"""
V2 External API - Credits Endpoints
Provides access to credit balance and transaction history.
"""
import logging
from datetime import datetime
from typing import Optional
from fastapi import APIRouter, Query, Security
from prisma.enums import APIKeyPermission
from pydantic import BaseModel, Field
from backend.api.external.middleware import require_permission
from backend.data.auth.base import APIAuthorizationInfo
from backend.data.credit import get_user_credit_model
from .common import DEFAULT_PAGE_SIZE, MAX_PAGE_SIZE
logger = logging.getLogger(__name__)
credits_router = APIRouter()
# ============================================================================
# Models
# ============================================================================
class CreditBalance(BaseModel):
"""User's credit balance."""
balance: int = Field(description="Current credit balance")
class CreditTransaction(BaseModel):
"""A credit transaction."""
transaction_key: str
amount: int = Field(description="Transaction amount (positive or negative)")
type: str = Field(description="One of: TOP_UP, USAGE, GRANT, REFUND")
transaction_time: datetime
running_balance: Optional[int] = Field(
default=None, description="Balance after this transaction"
)
description: Optional[str] = None
class CreditTransactionsResponse(BaseModel):
"""Response for listing credit transactions."""
transactions: list[CreditTransaction]
total_count: int
page: int
page_size: int
total_pages: int
# ============================================================================
# Endpoints
# ============================================================================
@credits_router.get(
path="",
summary="Get credit balance",
response_model=CreditBalance,
)
async def get_balance(
auth: APIAuthorizationInfo = Security(
require_permission(APIKeyPermission.READ_CREDITS)
),
) -> CreditBalance:
"""
Get the current credit balance for the authenticated user.
"""
user_credit_model = await get_user_credit_model(auth.user_id)
balance = await user_credit_model.get_credits(auth.user_id)
return CreditBalance(balance=balance)
@credits_router.get(
path="/transactions",
summary="Get transaction history",
response_model=CreditTransactionsResponse,
)
async def get_transactions(
auth: APIAuthorizationInfo = Security(
require_permission(APIKeyPermission.READ_CREDITS)
),
page: int = Query(default=1, ge=1, description="Page number (1-indexed)"),
page_size: int = Query(
default=DEFAULT_PAGE_SIZE,
ge=1,
le=MAX_PAGE_SIZE,
description=f"Items per page (max {MAX_PAGE_SIZE})",
),
transaction_type: Optional[str] = Query(
default=None,
description="Filter by transaction type (TOP_UP, USAGE, GRANT, REFUND)",
),
) -> CreditTransactionsResponse:
"""
Get credit transaction history for the authenticated user.
Returns transactions sorted by most recent first.
"""
user_credit_model = await get_user_credit_model(auth.user_id)
history = await user_credit_model.get_transaction_history(
user_id=auth.user_id,
transaction_count_limit=page_size,
transaction_type=transaction_type,
)
transactions = [
CreditTransaction(
transaction_key=t.transaction_key,
amount=t.amount,
type=t.transaction_type.value,
transaction_time=t.transaction_time,
running_balance=t.running_balance,
description=t.description,
)
for t in history.transactions
]
# Note: The current credit module doesn't support true pagination,
# so we're returning what we have
total_count = len(transactions)
total_pages = 1 # Without true pagination support
return CreditTransactionsResponse(
transactions=transactions,
total_count=total_count,
page=page,
page_size=page_size,
total_pages=total_pages,
)

View File

@@ -0,0 +1,132 @@
"""
V2 External API - Files Endpoints
Provides file upload functionality for agent inputs.
"""
import base64
import logging
from fastapi import APIRouter, File, HTTPException, Query, Security, UploadFile
from prisma.enums import APIKeyPermission
from pydantic import BaseModel, Field
from backend.api.external.middleware import require_permission
from backend.data.auth.base import APIAuthorizationInfo
from backend.util.cloud_storage import get_cloud_storage_handler
from backend.util.settings import Settings
from backend.util.virus_scanner import scan_content_safe
logger = logging.getLogger(__name__)
settings = Settings()
files_router = APIRouter()
# ============================================================================
# Models
# ============================================================================
class UploadFileResponse(BaseModel):
"""Response after uploading a file."""
file_uri: str = Field(description="URI to reference the uploaded file in agents")
file_name: str
size: int = Field(description="File size in bytes")
content_type: str
expires_in_hours: int
# ============================================================================
# Endpoints
# ============================================================================
def _create_file_size_error(size_bytes: int, max_size_mb: int) -> HTTPException:
"""Create standardized file size error response."""
return HTTPException(
status_code=400,
detail=f"File size ({size_bytes} bytes) exceeds the maximum allowed size of {max_size_mb}MB",
)
@files_router.post(
path="/upload",
summary="Upload a file",
response_model=UploadFileResponse,
)
async def upload_file(
file: UploadFile = File(...),
auth: APIAuthorizationInfo = Security(
require_permission(APIKeyPermission.UPLOAD_FILES)
),
provider: str = Query(
default="gcs", description="Storage provider (gcs, s3, azure)"
),
expiration_hours: int = Query(
default=24, ge=1, le=48, description="Hours until file expires (1-48)"
),
) -> UploadFileResponse:
"""
Upload a file to cloud storage for use with agents.
The returned `file_uri` can be used as input to agents that accept file inputs
(e.g., FileStoreBlock, AgentFileInputBlock).
Files are automatically scanned for viruses before storage.
"""
# Check file size limit
max_size_mb = settings.config.upload_file_size_limit_mb
max_size_bytes = max_size_mb * 1024 * 1024
# Try to get file size from headers first
if hasattr(file, "size") and file.size is not None and file.size > max_size_bytes:
raise _create_file_size_error(file.size, max_size_mb)
# Read file content
content = await file.read()
content_size = len(content)
# Double-check file size after reading
if content_size > max_size_bytes:
raise _create_file_size_error(content_size, max_size_mb)
# Extract file info
file_name = file.filename or "uploaded_file"
content_type = file.content_type or "application/octet-stream"
# Virus scan the content
await scan_content_safe(content, filename=file_name)
# Check if cloud storage is configured
cloud_storage = await get_cloud_storage_handler()
if not cloud_storage.config.gcs_bucket_name:
# Fallback to base64 data URI when GCS is not configured
base64_content = base64.b64encode(content).decode("utf-8")
data_uri = f"data:{content_type};base64,{base64_content}"
return UploadFileResponse(
file_uri=data_uri,
file_name=file_name,
size=content_size,
content_type=content_type,
expires_in_hours=expiration_hours,
)
# Store in cloud storage
storage_path = await cloud_storage.store_file(
content=content,
filename=file_name,
provider=provider,
expiration_hours=expiration_hours,
user_id=auth.user_id,
)
return UploadFileResponse(
file_uri=storage_path,
file_name=file_name,
size=content_size,
content_type=content_type,
expires_in_hours=expiration_hours,
)

View File

@@ -0,0 +1,445 @@
"""
V2 External API - Graphs Endpoints
Provides endpoints for managing agent graphs (CRUD operations).
"""
import logging
from fastapi import APIRouter, HTTPException, Query, Security
from prisma.enums import APIKeyPermission
from backend.api.external.middleware import require_permission
from backend.data import graph as graph_db
from backend.data.auth.base import APIAuthorizationInfo
from backend.integrations.webhooks.graph_lifecycle_hooks import (
on_graph_activate,
on_graph_deactivate,
)
from .common import DEFAULT_PAGE_SIZE, MAX_PAGE_SIZE
from .models import (
CreateGraphRequest,
DeleteGraphResponse,
GraphDetails,
GraphLink,
GraphMeta,
GraphNode,
GraphSettings,
GraphsListResponse,
SetActiveVersionRequest,
)
logger = logging.getLogger(__name__)
graphs_router = APIRouter()
def _convert_graph_meta(graph: graph_db.GraphMeta) -> GraphMeta:
"""Convert internal GraphMeta to v2 API model."""
return GraphMeta(
id=graph.id,
version=graph.version,
is_active=graph.is_active,
name=graph.name,
description=graph.description,
created_at=graph.created_at,
input_schema=graph.input_schema,
output_schema=graph.output_schema,
)
def _convert_graph_details(graph: graph_db.GraphModel) -> GraphDetails:
"""Convert internal GraphModel to v2 API GraphDetails model."""
return GraphDetails(
id=graph.id,
version=graph.version,
is_active=graph.is_active,
name=graph.name,
description=graph.description,
created_at=graph.created_at,
input_schema=graph.input_schema,
output_schema=graph.output_schema,
nodes=[
GraphNode(
id=node.id,
block_id=node.block_id,
input_default=node.input_default,
metadata=node.metadata,
)
for node in graph.nodes
],
links=[
GraphLink(
id=link.id,
source_id=link.source_id,
sink_id=link.sink_id,
source_name=link.source_name,
sink_name=link.sink_name,
is_static=link.is_static,
)
for link in graph.links
],
credentials_input_schema=graph.credentials_input_schema,
)
@graphs_router.get(
path="",
summary="List user's graphs",
response_model=GraphsListResponse,
)
async def list_graphs(
auth: APIAuthorizationInfo = Security(
require_permission(APIKeyPermission.READ_GRAPH)
),
page: int = Query(default=1, ge=1, description="Page number (1-indexed)"),
page_size: int = Query(
default=DEFAULT_PAGE_SIZE,
ge=1,
le=MAX_PAGE_SIZE,
description=f"Items per page (max {MAX_PAGE_SIZE})",
),
) -> GraphsListResponse:
"""
List all graphs owned by the authenticated user.
Returns a paginated list of graph metadata (not full graph details).
"""
graphs, pagination_info = await graph_db.list_graphs_paginated(
user_id=auth.user_id,
page=page,
page_size=page_size,
filter_by="active",
)
return GraphsListResponse(
graphs=[_convert_graph_meta(g) for g in graphs],
total_count=pagination_info.total_items,
page=pagination_info.current_page,
page_size=pagination_info.page_size,
total_pages=pagination_info.total_pages,
)
@graphs_router.post(
path="",
summary="Create a new graph",
response_model=GraphDetails,
)
async def create_graph(
create_graph_request: CreateGraphRequest,
auth: APIAuthorizationInfo = Security(
require_permission(APIKeyPermission.WRITE_GRAPH)
),
) -> GraphDetails:
"""
Create a new agent graph.
The graph will be validated and assigned a new ID. It will automatically
be added to the user's library.
"""
# Import here to avoid circular imports
from backend.api.features.library import db as library_db
# Convert v2 API Graph model to internal Graph model
internal_graph = graph_db.Graph(
id=create_graph_request.graph.id or "",
version=create_graph_request.graph.version,
is_active=create_graph_request.graph.is_active,
name=create_graph_request.graph.name,
description=create_graph_request.graph.description,
nodes=[
graph_db.Node(
id=node.id,
block_id=node.block_id,
input_default=node.input_default,
metadata=node.metadata,
)
for node in create_graph_request.graph.nodes
],
links=[
graph_db.Link(
id=link.id,
source_id=link.source_id,
sink_id=link.sink_id,
source_name=link.source_name,
sink_name=link.sink_name,
is_static=link.is_static,
)
for link in create_graph_request.graph.links
],
)
graph = graph_db.make_graph_model(internal_graph, auth.user_id)
graph.reassign_ids(user_id=auth.user_id, reassign_graph_id=True)
graph.validate_graph(for_run=False)
await graph_db.create_graph(graph, user_id=auth.user_id)
await library_db.create_library_agent(graph, user_id=auth.user_id)
activated_graph = await on_graph_activate(graph, user_id=auth.user_id)
return _convert_graph_details(activated_graph)
@graphs_router.get(
path="/{graph_id}",
summary="Get graph details",
response_model=GraphDetails,
)
async def get_graph(
graph_id: str,
auth: APIAuthorizationInfo = Security(
require_permission(APIKeyPermission.READ_GRAPH)
),
version: int | None = Query(
default=None,
description="Specific version to retrieve (default: active version)",
),
) -> GraphDetails:
"""
Get detailed information about a specific graph.
By default returns the active version. Use the `version` query parameter
to retrieve a specific version.
"""
graph = await graph_db.get_graph(
graph_id,
version,
user_id=auth.user_id,
include_subgraphs=True,
)
if not graph:
raise HTTPException(status_code=404, detail=f"Graph #{graph_id} not found.")
return _convert_graph_details(graph)
@graphs_router.put(
path="/{graph_id}",
summary="Update graph (creates new version)",
response_model=GraphDetails,
)
async def update_graph(
graph_id: str,
graph_request: CreateGraphRequest,
auth: APIAuthorizationInfo = Security(
require_permission(APIKeyPermission.WRITE_GRAPH)
),
) -> GraphDetails:
"""
Update a graph by creating a new version.
This does not modify existing versions - it creates a new version with
the provided content. The new version becomes the active version.
"""
# Import here to avoid circular imports
from backend.api.features.library import db as library_db
graph_data = graph_request.graph
if graph_data.id and graph_data.id != graph_id:
raise HTTPException(400, detail="Graph ID does not match ID in URI")
existing_versions = await graph_db.get_graph_all_versions(
graph_id, user_id=auth.user_id
)
if not existing_versions:
raise HTTPException(404, detail=f"Graph #{graph_id} not found")
latest_version_number = max(g.version for g in existing_versions)
# Convert v2 API Graph model to internal Graph model
internal_graph = graph_db.Graph(
id=graph_id,
version=latest_version_number + 1,
is_active=graph_data.is_active,
name=graph_data.name,
description=graph_data.description,
nodes=[
graph_db.Node(
id=node.id,
block_id=node.block_id,
input_default=node.input_default,
metadata=node.metadata,
)
for node in graph_data.nodes
],
links=[
graph_db.Link(
id=link.id,
source_id=link.source_id,
sink_id=link.sink_id,
source_name=link.source_name,
sink_name=link.sink_name,
is_static=link.is_static,
)
for link in graph_data.links
],
)
current_active_version = next((v for v in existing_versions if v.is_active), None)
graph = graph_db.make_graph_model(internal_graph, auth.user_id)
graph.reassign_ids(user_id=auth.user_id, reassign_graph_id=False)
graph.validate_graph(for_run=False)
new_graph_version = await graph_db.create_graph(graph, user_id=auth.user_id)
if new_graph_version.is_active:
await library_db.update_agent_version_in_library(
auth.user_id, new_graph_version.id, new_graph_version.version
)
new_graph_version = await on_graph_activate(
new_graph_version, user_id=auth.user_id
)
await graph_db.set_graph_active_version(
graph_id=graph_id, version=new_graph_version.version, user_id=auth.user_id
)
if current_active_version:
await on_graph_deactivate(current_active_version, user_id=auth.user_id)
new_graph_version_with_subgraphs = await graph_db.get_graph(
graph_id,
new_graph_version.version,
user_id=auth.user_id,
include_subgraphs=True,
)
assert new_graph_version_with_subgraphs
return _convert_graph_details(new_graph_version_with_subgraphs)
@graphs_router.delete(
path="/{graph_id}",
summary="Delete graph permanently",
response_model=DeleteGraphResponse,
)
async def delete_graph(
graph_id: str,
auth: APIAuthorizationInfo = Security(
require_permission(APIKeyPermission.WRITE_GRAPH)
),
) -> DeleteGraphResponse:
"""
Permanently delete a graph and all its versions.
This action cannot be undone. All associated executions will remain
but will reference a deleted graph.
"""
if active_version := await graph_db.get_graph(
graph_id=graph_id, version=None, user_id=auth.user_id
):
await on_graph_deactivate(active_version, user_id=auth.user_id)
version_count = await graph_db.delete_graph(graph_id, user_id=auth.user_id)
return DeleteGraphResponse(version_count=version_count)
@graphs_router.get(
path="/{graph_id}/versions",
summary="List all graph versions",
response_model=list[GraphDetails],
)
async def list_graph_versions(
graph_id: str,
auth: APIAuthorizationInfo = Security(
require_permission(APIKeyPermission.READ_GRAPH)
),
) -> list[GraphDetails]:
"""
Get all versions of a specific graph.
Returns a list of all versions, with the active version marked.
"""
graphs = await graph_db.get_graph_all_versions(graph_id, user_id=auth.user_id)
if not graphs:
raise HTTPException(status_code=404, detail=f"Graph #{graph_id} not found.")
return [_convert_graph_details(g) for g in graphs]
@graphs_router.put(
path="/{graph_id}/versions/active",
summary="Set active graph version",
)
async def set_active_version(
graph_id: str,
request_body: SetActiveVersionRequest,
auth: APIAuthorizationInfo = Security(
require_permission(APIKeyPermission.WRITE_GRAPH)
),
) -> None:
"""
Set which version of a graph is the active version.
The active version is used when executing the graph without specifying
a version number.
"""
# Import here to avoid circular imports
from backend.api.features.library import db as library_db
new_active_version = request_body.active_graph_version
new_active_graph = await graph_db.get_graph(
graph_id, new_active_version, user_id=auth.user_id
)
if not new_active_graph:
raise HTTPException(404, f"Graph #{graph_id} v{new_active_version} not found")
current_active_graph = await graph_db.get_graph(
graph_id=graph_id,
version=None,
user_id=auth.user_id,
)
await on_graph_activate(new_active_graph, user_id=auth.user_id)
await graph_db.set_graph_active_version(
graph_id=graph_id,
version=new_active_version,
user_id=auth.user_id,
)
await library_db.update_agent_version_in_library(
auth.user_id, new_active_graph.id, new_active_graph.version
)
if current_active_graph and current_active_graph.version != new_active_version:
await on_graph_deactivate(current_active_graph, user_id=auth.user_id)
@graphs_router.patch(
path="/{graph_id}/settings",
summary="Update graph settings",
response_model=GraphSettings,
)
async def update_graph_settings(
graph_id: str,
settings: GraphSettings,
auth: APIAuthorizationInfo = Security(
require_permission(APIKeyPermission.WRITE_GRAPH)
),
) -> GraphSettings:
"""
Update settings for a graph.
Currently supports:
- human_in_the_loop_safe_mode: Enable/disable safe mode for human-in-the-loop blocks
"""
# Import here to avoid circular imports
from backend.api.features.library import db as library_db
from backend.data.graph import GraphSettings as InternalGraphSettings
library_agent = await library_db.get_library_agent_by_graph_id(
graph_id=graph_id, user_id=auth.user_id
)
if not library_agent:
raise HTTPException(404, f"Graph #{graph_id} not found in user's library")
# Convert to internal model
internal_settings = InternalGraphSettings(
human_in_the_loop_safe_mode=settings.human_in_the_loop_safe_mode
)
updated_agent = await library_db.update_library_agent_settings(
user_id=auth.user_id,
agent_id=library_agent.id,
settings=internal_settings,
)
return GraphSettings(
human_in_the_loop_safe_mode=updated_agent.settings.human_in_the_loop_safe_mode
)

View File

@@ -0,0 +1,271 @@
"""
V2 External API - Integrations Endpoints
Provides access to user's integration credentials.
"""
import logging
from typing import Optional
from fastapi import APIRouter, HTTPException, Path, Security
from prisma.enums import APIKeyPermission
from pydantic import BaseModel, Field
from backend.api.external.middleware import require_permission
from backend.api.features.library import db as library_db
from backend.data import graph as graph_db
from backend.data.auth.base import APIAuthorizationInfo
from backend.data.model import Credentials, OAuth2Credentials
from backend.integrations.creds_manager import IntegrationCredentialsManager
logger = logging.getLogger(__name__)
integrations_router = APIRouter()
creds_manager = IntegrationCredentialsManager()
# ============================================================================
# Models
# ============================================================================
class Credential(BaseModel):
"""A user's credential for an integration."""
id: str
provider: str = Field(description="Integration provider name")
title: Optional[str] = Field(
default=None, description="User-assigned title for this credential"
)
scopes: list[str] = Field(default_factory=list, description="Granted scopes")
class CredentialsListResponse(BaseModel):
"""Response for listing credentials."""
credentials: list[Credential]
class CredentialRequirement(BaseModel):
"""A credential requirement for a graph or agent."""
provider: str = Field(description="Required provider name")
required_scopes: list[str] = Field(
default_factory=list, description="Required scopes"
)
matching_credentials: list[Credential] = Field(
default_factory=list,
description="User's credentials that match this requirement",
)
class CredentialRequirementsResponse(BaseModel):
"""Response for listing credential requirements."""
requirements: list[CredentialRequirement]
# ============================================================================
# Conversion Functions
# ============================================================================
def _convert_credential(cred: Credentials) -> Credential:
"""Convert internal credential to v2 API model."""
scopes: list[str] = []
if isinstance(cred, OAuth2Credentials):
scopes = cred.scopes or []
return Credential(
id=cred.id,
provider=cred.provider,
title=cred.title,
scopes=scopes,
)
# ============================================================================
# Endpoints
# ============================================================================
@integrations_router.get(
path="/credentials",
summary="List all credentials",
response_model=CredentialsListResponse,
)
async def list_credentials(
auth: APIAuthorizationInfo = Security(
require_permission(APIKeyPermission.READ_INTEGRATIONS)
),
) -> CredentialsListResponse:
"""
List all integration credentials for the authenticated user.
This returns all OAuth credentials the user has connected, across
all integration providers.
"""
credentials = await creds_manager.store.get_all_creds(auth.user_id)
return CredentialsListResponse(
credentials=[_convert_credential(c) for c in credentials]
)
@integrations_router.get(
path="/credentials/{provider}",
summary="List credentials by provider",
response_model=CredentialsListResponse,
)
async def list_credentials_by_provider(
provider: str = Path(description="Provider name (e.g., 'github', 'google')"),
auth: APIAuthorizationInfo = Security(
require_permission(APIKeyPermission.READ_INTEGRATIONS)
),
) -> CredentialsListResponse:
"""
List integration credentials for a specific provider.
"""
all_credentials = await creds_manager.store.get_all_creds(auth.user_id)
# Filter by provider
filtered = [c for c in all_credentials if c.provider.lower() == provider.lower()]
return CredentialsListResponse(
credentials=[_convert_credential(c) for c in filtered]
)
@integrations_router.get(
path="/graphs/{graph_id}/credentials",
summary="List credentials matching graph requirements",
response_model=CredentialRequirementsResponse,
)
async def list_graph_credential_requirements(
graph_id: str = Path(description="Graph ID"),
auth: APIAuthorizationInfo = Security(
require_permission(APIKeyPermission.READ_INTEGRATIONS)
),
) -> CredentialRequirementsResponse:
"""
List credential requirements for a graph and matching user credentials.
This helps identify which credentials the user needs to provide
when executing a graph.
"""
# Get the graph
graph = await graph_db.get_graph(
graph_id=graph_id,
version=None, # Active version
user_id=auth.user_id,
include_subgraphs=True,
)
if not graph:
raise HTTPException(status_code=404, detail=f"Graph #{graph_id} not found")
# Get the credentials input schema which contains provider requirements
creds_schema = graph.credentials_input_schema
all_credentials = await creds_manager.store.get_all_creds(auth.user_id)
requirements = []
for field_name, field_schema in creds_schema.get("properties", {}).items():
# Extract provider from schema
# The schema structure varies, but typically has provider info
providers = []
if "anyOf" in field_schema:
for option in field_schema["anyOf"]:
if "provider" in option:
providers.append(option["provider"])
elif "provider" in field_schema:
providers.append(field_schema["provider"])
for provider in providers:
# Find matching credentials
matching = [
_convert_credential(c)
for c in all_credentials
if c.provider.lower() == provider.lower()
]
requirements.append(
CredentialRequirement(
provider=provider,
required_scopes=[], # Would need to extract from schema
matching_credentials=matching,
)
)
return CredentialRequirementsResponse(requirements=requirements)
@integrations_router.get(
path="/library/{agent_id}/credentials",
summary="List credentials matching library agent requirements",
response_model=CredentialRequirementsResponse,
)
async def list_library_agent_credential_requirements(
agent_id: str = Path(description="Library agent ID"),
auth: APIAuthorizationInfo = Security(
require_permission(APIKeyPermission.READ_INTEGRATIONS)
),
) -> CredentialRequirementsResponse:
"""
List credential requirements for a library agent and matching user credentials.
This helps identify which credentials the user needs to provide
when executing an agent from their library.
"""
# Get the library agent
try:
library_agent = await library_db.get_library_agent(
id=agent_id,
user_id=auth.user_id,
)
except Exception:
raise HTTPException(status_code=404, detail=f"Agent #{agent_id} not found")
# Get the underlying graph
graph = await graph_db.get_graph(
graph_id=library_agent.graph_id,
version=library_agent.graph_version,
user_id=auth.user_id,
include_subgraphs=True,
)
if not graph:
raise HTTPException(
status_code=404,
detail=f"Graph for agent #{agent_id} not found",
)
# Get the credentials input schema
creds_schema = graph.credentials_input_schema
all_credentials = await creds_manager.store.get_all_creds(auth.user_id)
requirements = []
for field_name, field_schema in creds_schema.get("properties", {}).items():
# Extract provider from schema
providers = []
if "anyOf" in field_schema:
for option in field_schema["anyOf"]:
if "provider" in option:
providers.append(option["provider"])
elif "provider" in field_schema:
providers.append(field_schema["provider"])
for provider in providers:
# Find matching credentials
matching = [
_convert_credential(c)
for c in all_credentials
if c.provider.lower() == provider.lower()
]
requirements.append(
CredentialRequirement(
provider=provider,
required_scopes=[],
matching_credentials=matching,
)
)
return CredentialRequirementsResponse(requirements=requirements)

View File

@@ -0,0 +1,247 @@
"""
V2 External API - Library Endpoints
Provides access to the user's agent library and agent execution.
"""
import logging
from fastapi import APIRouter, HTTPException, Path, Query, Security
from prisma.enums import APIKeyPermission
from backend.api.external.middleware import require_permission
from backend.api.features.library import db as library_db
from backend.api.features.library import model as library_model
from backend.data import execution as execution_db
from backend.data.auth.base import APIAuthorizationInfo
from backend.data.credit import get_user_credit_model
from backend.executor import utils as execution_utils
from .common import DEFAULT_PAGE_SIZE, MAX_PAGE_SIZE
from .models import (
ExecuteAgentRequest,
LibraryAgent,
LibraryAgentsResponse,
Run,
RunsListResponse,
)
logger = logging.getLogger(__name__)
library_router = APIRouter()
# ============================================================================
# Conversion Functions
# ============================================================================
def _convert_library_agent(agent: library_model.LibraryAgent) -> LibraryAgent:
"""Convert internal LibraryAgent to v2 API model."""
return LibraryAgent(
id=agent.id,
graph_id=agent.graph_id,
graph_version=agent.graph_version,
name=agent.name,
description=agent.description,
is_favorite=agent.is_favorite,
can_access_graph=agent.can_access_graph,
is_latest_version=agent.is_latest_version,
image_url=agent.image_url,
creator_name=agent.creator_name,
input_schema=agent.input_schema,
output_schema=agent.output_schema,
created_at=agent.created_at,
updated_at=agent.updated_at,
)
def _convert_execution_to_run(exec: execution_db.GraphExecutionMeta) -> Run:
"""Convert internal execution to v2 API Run model."""
return Run(
id=exec.id,
graph_id=exec.graph_id,
graph_version=exec.graph_version,
status=exec.status.value,
started_at=exec.started_at,
ended_at=exec.ended_at,
inputs=exec.inputs,
cost=exec.stats.cost if exec.stats else 0,
duration=exec.stats.duration if exec.stats else 0,
node_count=exec.stats.node_exec_count if exec.stats else 0,
)
# ============================================================================
# Endpoints
# ============================================================================
@library_router.get(
path="/agents",
summary="List library agents",
response_model=LibraryAgentsResponse,
)
async def list_library_agents(
auth: APIAuthorizationInfo = Security(
require_permission(APIKeyPermission.READ_LIBRARY)
),
page: int = Query(default=1, ge=1, description="Page number (1-indexed)"),
page_size: int = Query(
default=DEFAULT_PAGE_SIZE,
ge=1,
le=MAX_PAGE_SIZE,
description=f"Items per page (max {MAX_PAGE_SIZE})",
),
) -> LibraryAgentsResponse:
"""
List agents in the user's library.
The library contains agents the user has created or added from the marketplace.
"""
result = await library_db.list_library_agents(
user_id=auth.user_id,
page=page,
page_size=page_size,
)
return LibraryAgentsResponse(
agents=[_convert_library_agent(a) for a in result.agents],
total_count=result.pagination.total_items,
page=result.pagination.current_page,
page_size=result.pagination.page_size,
total_pages=result.pagination.total_pages,
)
@library_router.get(
path="/agents/favorites",
summary="List favorite agents",
response_model=LibraryAgentsResponse,
)
async def list_favorite_agents(
auth: APIAuthorizationInfo = Security(
require_permission(APIKeyPermission.READ_LIBRARY)
),
page: int = Query(default=1, ge=1, description="Page number (1-indexed)"),
page_size: int = Query(
default=DEFAULT_PAGE_SIZE,
ge=1,
le=MAX_PAGE_SIZE,
description=f"Items per page (max {MAX_PAGE_SIZE})",
),
) -> LibraryAgentsResponse:
"""
List favorite agents in the user's library.
"""
result = await library_db.list_favorite_library_agents(
user_id=auth.user_id,
page=page,
page_size=page_size,
)
return LibraryAgentsResponse(
agents=[_convert_library_agent(a) for a in result.agents],
total_count=result.pagination.total_items,
page=result.pagination.current_page,
page_size=result.pagination.page_size,
total_pages=result.pagination.total_pages,
)
@library_router.post(
path="/agents/{agent_id}/runs",
summary="Execute an agent",
response_model=Run,
)
async def execute_agent(
request: ExecuteAgentRequest,
agent_id: str = Path(description="Library agent ID"),
auth: APIAuthorizationInfo = Security(
require_permission(APIKeyPermission.RUN_AGENT)
),
) -> Run:
"""
Execute an agent from the library.
This creates a new run with the provided inputs. The run executes
asynchronously and you can poll the run status using GET /runs/{run_id}.
"""
# Check credit balance
user_credit_model = await get_user_credit_model(auth.user_id)
current_balance = await user_credit_model.get_credits(auth.user_id)
if current_balance <= 0:
raise HTTPException(
status_code=402,
detail="Insufficient balance to execute the agent. Please top up your account.",
)
# Get the library agent to find the graph ID and version
try:
library_agent = await library_db.get_library_agent(
id=agent_id,
user_id=auth.user_id,
)
except Exception:
raise HTTPException(status_code=404, detail=f"Agent #{agent_id} not found")
try:
result = await execution_utils.add_graph_execution(
graph_id=library_agent.graph_id,
user_id=auth.user_id,
inputs=request.inputs,
graph_version=library_agent.graph_version,
graph_credentials_inputs=request.credentials_inputs,
)
return _convert_execution_to_run(result)
except Exception as e:
logger.error(f"Failed to execute agent: {e}")
raise HTTPException(status_code=400, detail=str(e))
@library_router.get(
path="/agents/{agent_id}/runs",
summary="List runs for an agent",
response_model=RunsListResponse,
)
async def list_agent_runs(
agent_id: str = Path(description="Library agent ID"),
auth: APIAuthorizationInfo = Security(
require_permission(APIKeyPermission.READ_LIBRARY)
),
page: int = Query(default=1, ge=1, description="Page number (1-indexed)"),
page_size: int = Query(
default=DEFAULT_PAGE_SIZE,
ge=1,
le=MAX_PAGE_SIZE,
description=f"Items per page (max {MAX_PAGE_SIZE})",
),
) -> RunsListResponse:
"""
List execution runs for a specific agent.
"""
# Get the library agent to find the graph ID
try:
library_agent = await library_db.get_library_agent(
id=agent_id,
user_id=auth.user_id,
)
except Exception:
raise HTTPException(status_code=404, detail=f"Agent #{agent_id} not found")
result = await execution_db.get_graph_executions_paginated(
graph_id=library_agent.graph_id,
user_id=auth.user_id,
page=page,
page_size=page_size,
)
return RunsListResponse(
runs=[_convert_execution_to_run(e) for e in result.executions],
total_count=result.pagination.total_items,
page=result.pagination.current_page,
page_size=result.pagination.page_size,
total_pages=result.pagination.total_pages,
)

View File

@@ -0,0 +1,510 @@
"""
V2 External API - Marketplace Endpoints
Provides access to the agent marketplace (store).
"""
import logging
import urllib.parse
from datetime import datetime
from typing import Literal, Optional
from fastapi import APIRouter, HTTPException, Path, Query, Security
from prisma.enums import APIKeyPermission
from pydantic import BaseModel, Field
from backend.api.external.middleware import require_permission
from backend.api.features.store import cache as store_cache
from backend.api.features.store import db as store_db
from backend.api.features.store import model as store_model
from backend.data.auth.base import APIAuthorizationInfo
from .common import DEFAULT_PAGE_SIZE, MAX_PAGE_SIZE
logger = logging.getLogger(__name__)
marketplace_router = APIRouter()
# ============================================================================
# Models
# ============================================================================
class MarketplaceAgent(BaseModel):
"""An agent available in the marketplace."""
slug: str
name: str
description: str
sub_heading: str
creator: str
creator_avatar: str
runs: int = Field(default=0, description="Number of times this agent has been run")
rating: float = Field(default=0.0, description="Average rating")
image_url: str = Field(default="")
class MarketplaceAgentDetails(BaseModel):
"""Detailed information about a marketplace agent."""
store_listing_version_id: str
slug: str
name: str
description: str
sub_heading: str
instructions: Optional[str] = None
creator: str
creator_avatar: str
categories: list[str] = Field(default_factory=list)
runs: int = Field(default=0)
rating: float = Field(default=0.0)
image_urls: list[str] = Field(default_factory=list)
video_url: str = Field(default="")
versions: list[str] = Field(default_factory=list, description="Available versions")
agent_graph_versions: list[str] = Field(default_factory=list)
agent_graph_id: str
last_updated: datetime
class MarketplaceAgentsResponse(BaseModel):
"""Response for listing marketplace agents."""
agents: list[MarketplaceAgent]
total_count: int
page: int
page_size: int
total_pages: int
class MarketplaceCreator(BaseModel):
"""A creator on the marketplace."""
name: str
username: str
description: str
avatar_url: str
num_agents: int
agent_rating: float
agent_runs: int
is_featured: bool = False
class MarketplaceCreatorDetails(BaseModel):
"""Detailed information about a marketplace creator."""
name: str
username: str
description: str
avatar_url: str
agent_rating: float
agent_runs: int
top_categories: list[str] = Field(default_factory=list)
links: list[str] = Field(default_factory=list)
class MarketplaceCreatorsResponse(BaseModel):
"""Response for listing marketplace creators."""
creators: list[MarketplaceCreator]
total_count: int
page: int
page_size: int
total_pages: int
class MarketplaceSubmission(BaseModel):
"""A marketplace submission."""
graph_id: str
graph_version: int
name: str
sub_heading: str
slug: str
description: str
instructions: Optional[str] = None
image_urls: list[str] = Field(default_factory=list)
date_submitted: datetime
status: str = Field(description="One of: DRAFT, PENDING, APPROVED, REJECTED")
runs: int = Field(default=0)
rating: float = Field(default=0.0)
store_listing_version_id: Optional[str] = None
version: Optional[int] = None
review_comments: Optional[str] = None
reviewed_at: Optional[datetime] = None
video_url: Optional[str] = None
categories: list[str] = Field(default_factory=list)
class SubmissionsListResponse(BaseModel):
"""Response for listing submissions."""
submissions: list[MarketplaceSubmission]
total_count: int
page: int
page_size: int
total_pages: int
class CreateSubmissionRequest(BaseModel):
"""Request to create a marketplace submission."""
graph_id: str = Field(description="ID of the graph to submit")
graph_version: int = Field(description="Version of the graph to submit")
name: str = Field(description="Display name for the agent")
slug: str = Field(description="URL-friendly identifier")
description: str = Field(description="Full description")
sub_heading: str = Field(description="Short tagline")
image_urls: list[str] = Field(default_factory=list)
video_url: Optional[str] = None
categories: list[str] = Field(default_factory=list)
# ============================================================================
# Conversion Functions
# ============================================================================
def _convert_store_agent(agent: store_model.StoreAgent) -> MarketplaceAgent:
"""Convert internal StoreAgent to v2 API model."""
return MarketplaceAgent(
slug=agent.slug,
name=agent.agent_name,
description=agent.description,
sub_heading=agent.sub_heading,
creator=agent.creator,
creator_avatar=agent.creator_avatar,
runs=agent.runs,
rating=agent.rating,
image_url=agent.agent_image,
)
def _convert_store_agent_details(
agent: store_model.StoreAgentDetails,
) -> MarketplaceAgentDetails:
"""Convert internal StoreAgentDetails to v2 API model."""
return MarketplaceAgentDetails(
store_listing_version_id=agent.store_listing_version_id,
slug=agent.slug,
name=agent.agent_name,
description=agent.description,
sub_heading=agent.sub_heading,
instructions=agent.instructions,
creator=agent.creator,
creator_avatar=agent.creator_avatar,
categories=agent.categories,
runs=agent.runs,
rating=agent.rating,
image_urls=agent.agent_image,
video_url=agent.agent_video,
versions=agent.versions,
agent_graph_versions=agent.agentGraphVersions,
agent_graph_id=agent.agentGraphId,
last_updated=agent.last_updated,
)
def _convert_creator(creator: store_model.Creator) -> MarketplaceCreator:
"""Convert internal Creator to v2 API model."""
return MarketplaceCreator(
name=creator.name,
username=creator.username,
description=creator.description,
avatar_url=creator.avatar_url,
num_agents=creator.num_agents,
agent_rating=creator.agent_rating,
agent_runs=creator.agent_runs,
is_featured=creator.is_featured,
)
def _convert_creator_details(
creator: store_model.CreatorDetails,
) -> MarketplaceCreatorDetails:
"""Convert internal CreatorDetails to v2 API model."""
return MarketplaceCreatorDetails(
name=creator.name,
username=creator.username,
description=creator.description,
avatar_url=creator.avatar_url,
agent_rating=creator.agent_rating,
agent_runs=creator.agent_runs,
top_categories=creator.top_categories,
links=creator.links,
)
def _convert_submission(sub: store_model.StoreSubmission) -> MarketplaceSubmission:
"""Convert internal StoreSubmission to v2 API model."""
return MarketplaceSubmission(
graph_id=sub.agent_id,
graph_version=sub.agent_version,
name=sub.name,
sub_heading=sub.sub_heading,
slug=sub.slug,
description=sub.description,
instructions=sub.instructions,
image_urls=sub.image_urls,
date_submitted=sub.date_submitted,
status=sub.status.value,
runs=sub.runs,
rating=sub.rating,
store_listing_version_id=sub.store_listing_version_id,
version=sub.version,
review_comments=sub.review_comments,
reviewed_at=sub.reviewed_at,
video_url=sub.video_url,
categories=sub.categories,
)
# ============================================================================
# Endpoints - Read (authenticated)
# ============================================================================
@marketplace_router.get(
path="/agents",
summary="List marketplace agents",
response_model=MarketplaceAgentsResponse,
)
async def list_agents(
auth: APIAuthorizationInfo = Security(
require_permission(APIKeyPermission.READ_STORE)
),
featured: bool = Query(default=False, description="Filter to featured agents only"),
creator: Optional[str] = Query(
default=None, description="Filter by creator username"
),
sorted_by: Optional[Literal["rating", "runs", "name", "updated_at"]] = Query(
default=None, description="Sort field"
),
search_query: Optional[str] = Query(default=None, description="Search query"),
category: Optional[str] = Query(default=None, description="Filter by category"),
page: int = Query(default=1, ge=1, description="Page number (1-indexed)"),
page_size: int = Query(
default=DEFAULT_PAGE_SIZE,
ge=1,
le=MAX_PAGE_SIZE,
description=f"Items per page (max {MAX_PAGE_SIZE})",
),
) -> MarketplaceAgentsResponse:
"""
List agents available in the marketplace.
Supports filtering by featured status, creator, category, and search query.
Results can be sorted by rating, runs, name, or update time.
"""
result = await store_cache._get_cached_store_agents(
featured=featured,
creator=creator,
sorted_by=sorted_by,
search_query=search_query,
category=category,
page=page,
page_size=page_size,
)
return MarketplaceAgentsResponse(
agents=[_convert_store_agent(a) for a in result.agents],
total_count=result.pagination.total_items,
page=result.pagination.current_page,
page_size=result.pagination.page_size,
total_pages=result.pagination.total_pages,
)
@marketplace_router.get(
path="/agents/{username}/{agent_name}",
summary="Get agent details",
response_model=MarketplaceAgentDetails,
)
async def get_agent_details(
username: str = Path(description="Creator username"),
agent_name: str = Path(description="Agent slug/name"),
auth: APIAuthorizationInfo = Security(
require_permission(APIKeyPermission.READ_STORE)
),
) -> MarketplaceAgentDetails:
"""
Get detailed information about a specific marketplace agent.
"""
username = urllib.parse.unquote(username).lower()
agent_name = urllib.parse.unquote(agent_name).lower()
agent = await store_cache._get_cached_agent_details(
username=username, agent_name=agent_name
)
return _convert_store_agent_details(agent)
@marketplace_router.get(
path="/creators",
summary="List marketplace creators",
response_model=MarketplaceCreatorsResponse,
)
async def list_creators(
auth: APIAuthorizationInfo = Security(
require_permission(APIKeyPermission.READ_STORE)
),
featured: bool = Query(
default=False, description="Filter to featured creators only"
),
search_query: Optional[str] = Query(default=None, description="Search query"),
sorted_by: Optional[Literal["agent_rating", "agent_runs", "num_agents"]] = Query(
default=None, description="Sort field"
),
page: int = Query(default=1, ge=1, description="Page number (1-indexed)"),
page_size: int = Query(
default=DEFAULT_PAGE_SIZE,
ge=1,
le=MAX_PAGE_SIZE,
description=f"Items per page (max {MAX_PAGE_SIZE})",
),
) -> MarketplaceCreatorsResponse:
"""
List creators on the marketplace.
Supports filtering by featured status and search query.
Results can be sorted by rating, runs, or number of agents.
"""
result = await store_cache._get_cached_store_creators(
featured=featured,
search_query=search_query,
sorted_by=sorted_by,
page=page,
page_size=page_size,
)
return MarketplaceCreatorsResponse(
creators=[_convert_creator(c) for c in result.creators],
total_count=result.pagination.total_items,
page=result.pagination.current_page,
page_size=result.pagination.page_size,
total_pages=result.pagination.total_pages,
)
@marketplace_router.get(
path="/creators/{username}",
summary="Get creator details",
response_model=MarketplaceCreatorDetails,
)
async def get_creator_details(
username: str = Path(description="Creator username"),
auth: APIAuthorizationInfo = Security(
require_permission(APIKeyPermission.READ_STORE)
),
) -> MarketplaceCreatorDetails:
"""
Get detailed information about a specific marketplace creator.
"""
username = urllib.parse.unquote(username).lower()
creator = await store_cache._get_cached_creator_details(username=username)
return _convert_creator_details(creator)
# ============================================================================
# Endpoints - Submissions (CRUD)
# ============================================================================
@marketplace_router.get(
path="/submissions",
summary="List my submissions",
response_model=SubmissionsListResponse,
)
async def list_submissions(
auth: APIAuthorizationInfo = Security(
require_permission(APIKeyPermission.READ_STORE)
),
page: int = Query(default=1, ge=1, description="Page number (1-indexed)"),
page_size: int = Query(
default=DEFAULT_PAGE_SIZE,
ge=1,
le=MAX_PAGE_SIZE,
description=f"Items per page (max {MAX_PAGE_SIZE})",
),
) -> SubmissionsListResponse:
"""
List your marketplace submissions.
Returns all submissions you've created, including drafts, pending,
approved, and rejected submissions.
"""
result = await store_db.get_store_submissions(
user_id=auth.user_id,
page=page,
page_size=page_size,
)
return SubmissionsListResponse(
submissions=[_convert_submission(s) for s in result.submissions],
total_count=result.pagination.total_items,
page=result.pagination.current_page,
page_size=result.pagination.page_size,
total_pages=result.pagination.total_pages,
)
@marketplace_router.post(
path="/submissions",
summary="Create a submission",
response_model=MarketplaceSubmission,
)
async def create_submission(
request: CreateSubmissionRequest,
auth: APIAuthorizationInfo = Security(
require_permission(APIKeyPermission.WRITE_STORE)
),
) -> MarketplaceSubmission:
"""
Create a new marketplace submission.
This submits an agent for review to be published in the marketplace.
The submission will be in PENDING status until reviewed by the team.
"""
submission = await store_db.create_store_submission(
user_id=auth.user_id,
agent_id=request.graph_id,
agent_version=request.graph_version,
slug=request.slug,
name=request.name,
sub_heading=request.sub_heading,
description=request.description,
image_urls=request.image_urls,
video_url=request.video_url,
categories=request.categories,
)
return _convert_submission(submission)
@marketplace_router.delete(
path="/submissions/{submission_id}",
summary="Delete a submission",
)
async def delete_submission(
submission_id: str = Path(description="Submission ID"),
auth: APIAuthorizationInfo = Security(
require_permission(APIKeyPermission.WRITE_STORE)
),
) -> None:
"""
Delete a marketplace submission.
Only submissions in DRAFT status can be deleted.
"""
success = await store_db.delete_store_submission(
user_id=auth.user_id,
submission_id=submission_id,
)
if not success:
raise HTTPException(
status_code=404, detail=f"Submission #{submission_id} not found"
)

View File

@@ -0,0 +1,552 @@
"""
V2 External API - Request and Response Models
This module defines all request and response models for the v2 external API.
All models are self-contained and specific to the external API contract.
"""
from datetime import datetime
from typing import Any, Optional
from pydantic import BaseModel, Field
# ============================================================================
# Common/Shared Models
# ============================================================================
class PaginatedResponse(BaseModel):
"""Base class for paginated responses."""
total_count: int = Field(description="Total number of items across all pages")
page: int = Field(description="Current page number (1-indexed)")
page_size: int = Field(description="Number of items per page")
total_pages: int = Field(description="Total number of pages")
# ============================================================================
# Graph Models
# ============================================================================
class GraphLink(BaseModel):
"""A link between two nodes in a graph."""
id: str
source_id: str = Field(description="ID of the source node")
sink_id: str = Field(description="ID of the target node")
source_name: str = Field(description="Output pin name on source node")
sink_name: str = Field(description="Input pin name on target node")
is_static: bool = Field(
default=False, description="Whether this link provides static data"
)
class GraphNode(BaseModel):
"""A node in an agent graph."""
id: str
block_id: str = Field(description="ID of the block type")
input_default: dict[str, Any] = Field(
default_factory=dict, description="Default input values"
)
metadata: dict[str, Any] = Field(
default_factory=dict, description="Node metadata (e.g., position)"
)
class Graph(BaseModel):
"""Graph definition for creating or updating an agent."""
id: Optional[str] = Field(default=None, description="Graph ID (assigned by server)")
version: int = Field(default=1, description="Graph version")
is_active: bool = Field(default=True, description="Whether this version is active")
name: str = Field(description="Graph name")
description: str = Field(default="", description="Graph description")
nodes: list[GraphNode] = Field(default_factory=list, description="List of nodes")
links: list[GraphLink] = Field(
default_factory=list, description="Links between nodes"
)
class GraphMeta(BaseModel):
"""Graph metadata (summary information)."""
id: str
version: int
is_active: bool
name: str
description: str
created_at: datetime
input_schema: dict[str, Any] = Field(description="Input schema for the graph")
output_schema: dict[str, Any] = Field(description="Output schema for the graph")
class GraphDetails(GraphMeta):
"""Full graph details including nodes and links."""
nodes: list[GraphNode]
links: list[GraphLink]
credentials_input_schema: dict[str, Any] = Field(
description="Schema for required credentials"
)
class GraphSettings(BaseModel):
"""Settings for a graph."""
human_in_the_loop_safe_mode: Optional[bool] = Field(
default=None, description="Enable safe mode for human-in-the-loop blocks"
)
class CreateGraphRequest(BaseModel):
"""Request to create a new graph."""
graph: Graph = Field(description="The graph definition")
class SetActiveVersionRequest(BaseModel):
"""Request to set the active graph version."""
active_graph_version: int = Field(description="Version number to set as active")
class GraphsListResponse(PaginatedResponse):
"""Response for listing graphs."""
graphs: list[GraphMeta]
class DeleteGraphResponse(BaseModel):
"""Response for deleting a graph."""
version_count: int = Field(description="Number of versions deleted")
# ============================================================================
# Schedule Models
# ============================================================================
class Schedule(BaseModel):
"""An execution schedule for a graph."""
id: str
name: str
graph_id: str
graph_version: int
cron: str = Field(description="Cron expression for the schedule")
input_data: dict[str, Any] = Field(
default_factory=dict, description="Input data for scheduled executions"
)
is_enabled: bool = Field(default=True, description="Whether schedule is enabled")
next_run_time: Optional[datetime] = Field(
default=None, description="Next scheduled run time"
)
class CreateScheduleRequest(BaseModel):
"""Request to create a schedule."""
name: str = Field(description="Display name for the schedule")
cron: str = Field(description="Cron expression (e.g., '0 9 * * *' for 9am daily)")
input_data: dict[str, Any] = Field(
default_factory=dict, description="Input data for scheduled executions"
)
credentials_inputs: dict[str, Any] = Field(
default_factory=dict, description="Credentials for the schedule"
)
graph_version: Optional[int] = Field(
default=None, description="Graph version (default: active version)"
)
timezone: Optional[str] = Field(
default=None,
description="Timezone for schedule (e.g., 'America/New_York')",
)
class SchedulesListResponse(PaginatedResponse):
"""Response for listing schedules."""
schedules: list[Schedule]
# ============================================================================
# Block Models
# ============================================================================
class BlockCost(BaseModel):
"""Cost information for a block."""
cost_type: str = Field(description="Type of cost (e.g., 'per_call', 'per_token')")
cost_filter: dict[str, Any] = Field(
default_factory=dict, description="Conditions for this cost"
)
cost_amount: int = Field(description="Cost amount in credits")
class Block(BaseModel):
"""A building block that can be used in graphs."""
id: str
name: str
description: str
categories: list[str] = Field(default_factory=list)
input_schema: dict[str, Any]
output_schema: dict[str, Any]
costs: list[BlockCost] = Field(default_factory=list)
class BlocksListResponse(BaseModel):
"""Response for listing blocks."""
blocks: list[Block]
# ============================================================================
# Marketplace Models
# ============================================================================
class MarketplaceAgent(BaseModel):
"""An agent available in the marketplace."""
slug: str
agent_name: str
agent_image: str
creator: str
creator_avatar: str
sub_heading: str
description: str
runs: int = Field(default=0, description="Number of times this agent has been run")
rating: float = Field(default=0.0, description="Average rating")
class MarketplaceAgentDetails(BaseModel):
"""Detailed information about a marketplace agent."""
store_listing_version_id: str
slug: str
agent_name: str
agent_video: str
agent_output_demo: str
agent_image: list[str]
creator: str
creator_avatar: str
sub_heading: str
description: str
instructions: Optional[str] = None
categories: list[str]
runs: int
rating: float
versions: list[str]
agent_graph_versions: list[str]
agent_graph_id: str
last_updated: datetime
recommended_schedule_cron: Optional[str] = None
class MarketplaceCreator(BaseModel):
"""A creator on the marketplace."""
name: str
username: str
description: str
avatar_url: str
num_agents: int
agent_rating: float
agent_runs: int
is_featured: bool = False
class MarketplaceAgentsResponse(PaginatedResponse):
"""Response for listing marketplace agents."""
agents: list[MarketplaceAgent]
class MarketplaceCreatorsResponse(PaginatedResponse):
"""Response for listing marketplace creators."""
creators: list[MarketplaceCreator]
# Submission models
class MarketplaceSubmission(BaseModel):
"""A marketplace submission."""
agent_id: str
agent_version: int
name: str
sub_heading: str
slug: str
description: str
instructions: Optional[str] = None
image_urls: list[str] = Field(default_factory=list)
date_submitted: datetime
status: str = Field(description="One of: DRAFT, PENDING, APPROVED, REJECTED")
runs: int
rating: float
store_listing_version_id: Optional[str] = None
version: Optional[int] = None
# Review fields
review_comments: Optional[str] = None
reviewed_at: Optional[datetime] = None
# Additional optional fields
video_url: Optional[str] = None
categories: list[str] = Field(default_factory=list)
class CreateSubmissionRequest(BaseModel):
"""Request to create a marketplace submission."""
agent_id: str = Field(description="ID of the graph to submit")
agent_version: int = Field(description="Version of the graph to submit")
name: str = Field(description="Display name for the agent")
slug: str = Field(description="URL-friendly identifier")
description: str = Field(description="Full description")
sub_heading: str = Field(description="Short tagline")
image_urls: list[str] = Field(default_factory=list)
video_url: Optional[str] = None
categories: list[str] = Field(default_factory=list)
class UpdateSubmissionRequest(BaseModel):
"""Request to update a marketplace submission."""
name: Optional[str] = None
description: Optional[str] = None
sub_heading: Optional[str] = None
image_urls: Optional[list[str]] = None
video_url: Optional[str] = None
categories: Optional[list[str]] = None
class SubmissionsListResponse(PaginatedResponse):
"""Response for listing submissions."""
submissions: list[MarketplaceSubmission]
# ============================================================================
# Library Models
# ============================================================================
class LibraryAgent(BaseModel):
"""An agent in the user's library."""
id: str
graph_id: str
graph_version: int
name: str
description: str
is_favorite: bool = False
can_access_graph: bool = False
is_latest_version: bool = False
image_url: Optional[str] = None
creator_name: str
input_schema: dict[str, Any] = Field(description="Input schema for the agent")
output_schema: dict[str, Any] = Field(description="Output schema for the agent")
created_at: datetime
updated_at: datetime
class LibraryAgentsResponse(PaginatedResponse):
"""Response for listing library agents."""
agents: list[LibraryAgent]
class ExecuteAgentRequest(BaseModel):
"""Request to execute an agent."""
inputs: dict[str, Any] = Field(
default_factory=dict, description="Input values for the agent"
)
credentials_inputs: dict[str, Any] = Field(
default_factory=dict, description="Credentials for the agent"
)
# ============================================================================
# Run Models
# ============================================================================
class Run(BaseModel):
"""An execution run."""
id: str
graph_id: str
graph_version: int
status: str = Field(
description="One of: INCOMPLETE, QUEUED, RUNNING, COMPLETED, TERMINATED, FAILED, REVIEW"
)
started_at: datetime
ended_at: Optional[datetime] = None
inputs: Optional[dict[str, Any]] = None
cost: int = Field(default=0, description="Cost in credits")
duration: float = Field(default=0, description="Duration in seconds")
node_count: int = Field(default=0, description="Number of nodes executed")
class RunDetails(Run):
"""Detailed information about a run including node executions."""
outputs: Optional[dict[str, list[Any]]] = None
node_executions: list[dict[str, Any]] = Field(
default_factory=list, description="Individual node execution results"
)
class RunsListResponse(PaginatedResponse):
"""Response for listing runs."""
runs: list[Run]
# ============================================================================
# Run Review Models (Human-in-the-loop)
# ============================================================================
class PendingReview(BaseModel):
"""A pending human-in-the-loop review."""
id: str # node_exec_id
run_id: str
graph_id: str
graph_version: int
payload: Any = Field(description="Data to be reviewed")
instructions: Optional[str] = Field(
default=None, description="Instructions for the reviewer"
)
editable: bool = Field(
default=True, description="Whether the reviewer can edit the data"
)
status: str = Field(description="One of: WAITING, APPROVED, REJECTED")
created_at: datetime
class PendingReviewsResponse(PaginatedResponse):
"""Response for listing pending reviews."""
reviews: list[PendingReview]
class ReviewDecision(BaseModel):
"""Decision for a single review item."""
node_exec_id: str = Field(description="Node execution ID (review ID)")
approved: bool = Field(description="Whether to approve the data")
edited_payload: Optional[Any] = Field(
default=None, description="Modified payload data (if editing)"
)
message: Optional[str] = Field(
default=None, description="Optional message from reviewer", max_length=2000
)
class SubmitReviewsRequest(BaseModel):
"""Request to submit review responses for all pending reviews of an execution."""
reviews: list[ReviewDecision] = Field(
description="All review decisions for the execution"
)
class SubmitReviewsResponse(BaseModel):
"""Response after submitting reviews."""
run_id: str
approved_count: int = Field(description="Number of reviews approved")
rejected_count: int = Field(description="Number of reviews rejected")
# ============================================================================
# Credit Models
# ============================================================================
class CreditBalance(BaseModel):
"""User's credit balance."""
balance: int = Field(description="Current credit balance")
class CreditTransaction(BaseModel):
"""A credit transaction."""
transaction_key: str
amount: int
transaction_type: str = Field(description="Transaction type")
transaction_time: datetime
running_balance: int
description: Optional[str] = None
class CreditTransactionsResponse(PaginatedResponse):
"""Response for listing credit transactions."""
transactions: list[CreditTransaction]
# ============================================================================
# Integration Models
# ============================================================================
class Credential(BaseModel):
"""A user's credential for an integration."""
id: str
provider: str = Field(description="Integration provider name")
title: Optional[str] = Field(
default=None, description="User-assigned title for this credential"
)
scopes: list[str] = Field(default_factory=list, description="Granted scopes")
class CredentialsListResponse(BaseModel):
"""Response for listing credentials."""
credentials: list[Credential]
class CredentialRequirement(BaseModel):
"""A credential requirement for a graph or agent."""
provider: str = Field(description="Required provider name")
required_scopes: list[str] = Field(
default_factory=list, description="Required scopes"
)
matching_credentials: list[Credential] = Field(
default_factory=list,
description="User's credentials that match this requirement",
)
class CredentialRequirementsResponse(BaseModel):
"""Response for listing credential requirements."""
requirements: list[CredentialRequirement]
# ============================================================================
# File Models
# ============================================================================
class UploadFileResponse(BaseModel):
"""Response after uploading a file."""
file_uri: str = Field(description="URI to reference the uploaded file")
file_name: str
size: int = Field(description="File size in bytes")
content_type: str
expires_in_hours: int

View File

@@ -0,0 +1,35 @@
"""
V2 External API Routes
This module defines the main v2 router that aggregates all v2 API endpoints.
"""
from fastapi import APIRouter
from .blocks import blocks_router
from .credits import credits_router
from .files import files_router
from .graphs import graphs_router
from .integrations import integrations_router
from .library import library_router
from .marketplace import marketplace_router
from .runs import runs_router
from .schedules import graph_schedules_router, schedules_router
v2_router = APIRouter()
# Include all sub-routers
v2_router.include_router(graphs_router, prefix="/graphs", tags=["graphs"])
v2_router.include_router(graph_schedules_router, prefix="/graphs", tags=["schedules"])
v2_router.include_router(schedules_router, prefix="/schedules", tags=["schedules"])
v2_router.include_router(blocks_router, prefix="/blocks", tags=["blocks"])
v2_router.include_router(
marketplace_router, prefix="/marketplace", tags=["marketplace"]
)
v2_router.include_router(library_router, prefix="/library", tags=["library"])
v2_router.include_router(runs_router, prefix="/runs", tags=["runs"])
v2_router.include_router(credits_router, prefix="/credits", tags=["credits"])
v2_router.include_router(
integrations_router, prefix="/integrations", tags=["integrations"]
)
v2_router.include_router(files_router, prefix="/files", tags=["files"])

View File

@@ -0,0 +1,451 @@
"""
V2 External API - Runs Endpoints
Provides access to execution runs and human-in-the-loop reviews.
"""
import logging
from datetime import datetime
from typing import Any, Optional
from fastapi import APIRouter, HTTPException, Path, Query, Security
from prisma.enums import APIKeyPermission, ReviewStatus
from pydantic import BaseModel, Field
from backend.api.external.middleware import require_permission
from backend.api.features.executions.review.model import (
PendingHumanReviewModel,
SafeJsonData,
)
from backend.data import execution as execution_db
from backend.data import human_review as review_db
from backend.data.auth.base import APIAuthorizationInfo
from backend.executor import utils as execution_utils
from .common import DEFAULT_PAGE_SIZE, MAX_PAGE_SIZE
logger = logging.getLogger(__name__)
runs_router = APIRouter()
# ============================================================================
# Models
# ============================================================================
class Run(BaseModel):
"""An execution run."""
id: str
graph_id: str
graph_version: int
status: str = Field(
description="One of: INCOMPLETE, QUEUED, RUNNING, COMPLETED, TERMINATED, FAILED, REVIEW"
)
started_at: datetime
ended_at: Optional[datetime] = None
inputs: Optional[dict[str, Any]] = None
cost: int = Field(default=0, description="Cost in credits")
duration: float = Field(default=0, description="Duration in seconds")
node_count: int = Field(default=0, description="Number of nodes executed")
class RunDetails(Run):
"""Detailed information about a run including outputs and node executions."""
outputs: Optional[dict[str, list[Any]]] = None
node_executions: list[dict[str, Any]] = Field(
default_factory=list, description="Individual node execution results"
)
class RunsListResponse(BaseModel):
"""Response for listing runs."""
runs: list[Run]
total_count: int
page: int
page_size: int
total_pages: int
class PendingReview(BaseModel):
"""A pending human-in-the-loop review."""
id: str # node_exec_id
run_id: str
graph_id: str
graph_version: int
payload: SafeJsonData = Field(description="Data to be reviewed")
instructions: Optional[str] = Field(
default=None, description="Instructions for the reviewer"
)
editable: bool = Field(
default=True, description="Whether the reviewer can edit the data"
)
status: str = Field(description="One of: WAITING, APPROVED, REJECTED")
created_at: datetime
class PendingReviewsResponse(BaseModel):
"""Response for listing pending reviews."""
reviews: list[PendingReview]
total_count: int
page: int
page_size: int
total_pages: int
class ReviewDecision(BaseModel):
"""Decision for a single review item."""
node_exec_id: str = Field(description="Node execution ID (review ID)")
approved: bool = Field(description="Whether to approve the data")
edited_payload: Optional[SafeJsonData] = Field(
default=None, description="Modified payload data (if editing)"
)
message: Optional[str] = Field(
default=None, description="Optional message from reviewer", max_length=2000
)
class SubmitReviewsRequest(BaseModel):
"""Request to submit review responses for all pending reviews of an execution."""
reviews: list[ReviewDecision] = Field(
description="All review decisions for the execution"
)
class SubmitReviewsResponse(BaseModel):
"""Response after submitting reviews."""
run_id: str
approved_count: int = Field(description="Number of reviews approved")
rejected_count: int = Field(description="Number of reviews rejected")
# ============================================================================
# Conversion Functions
# ============================================================================
def _convert_execution_to_run(exec: execution_db.GraphExecutionMeta) -> Run:
"""Convert internal execution to v2 API Run model."""
return Run(
id=exec.id,
graph_id=exec.graph_id,
graph_version=exec.graph_version,
status=exec.status.value,
started_at=exec.started_at,
ended_at=exec.ended_at,
inputs=exec.inputs,
cost=exec.stats.cost if exec.stats else 0,
duration=exec.stats.duration if exec.stats else 0,
node_count=exec.stats.node_exec_count if exec.stats else 0,
)
def _convert_execution_to_run_details(
exec: execution_db.GraphExecutionWithNodes,
) -> RunDetails:
"""Convert internal execution with nodes to v2 API RunDetails model."""
return RunDetails(
id=exec.id,
graph_id=exec.graph_id,
graph_version=exec.graph_version,
status=exec.status.value,
started_at=exec.started_at,
ended_at=exec.ended_at,
inputs=exec.inputs,
outputs=exec.outputs,
cost=exec.stats.cost if exec.stats else 0,
duration=exec.stats.duration if exec.stats else 0,
node_count=exec.stats.node_exec_count if exec.stats else 0,
node_executions=[
{
"node_id": node.node_id,
"status": node.status.value,
"input_data": node.input_data,
"output_data": node.output_data,
"started_at": node.start_time,
"ended_at": node.end_time,
}
for node in exec.node_executions
],
)
def _convert_pending_review(review: PendingHumanReviewModel) -> PendingReview:
"""Convert internal PendingHumanReviewModel to v2 API PendingReview model."""
return PendingReview(
id=review.node_exec_id,
run_id=review.graph_exec_id,
graph_id=review.graph_id,
graph_version=review.graph_version,
payload=review.payload,
instructions=review.instructions,
editable=review.editable,
status=review.status.value,
created_at=review.created_at,
)
# ============================================================================
# Endpoints - Runs
# ============================================================================
@runs_router.get(
path="",
summary="List all runs",
response_model=RunsListResponse,
)
async def list_runs(
auth: APIAuthorizationInfo = Security(
require_permission(APIKeyPermission.READ_RUN)
),
page: int = Query(default=1, ge=1, description="Page number (1-indexed)"),
page_size: int = Query(
default=DEFAULT_PAGE_SIZE,
ge=1,
le=MAX_PAGE_SIZE,
description=f"Items per page (max {MAX_PAGE_SIZE})",
),
) -> RunsListResponse:
"""
List all execution runs for the authenticated user.
Returns runs across all agents, sorted by most recent first.
"""
result = await execution_db.get_graph_executions_paginated(
user_id=auth.user_id,
page=page,
page_size=page_size,
)
return RunsListResponse(
runs=[_convert_execution_to_run(e) for e in result.executions],
total_count=result.pagination.total_items,
page=result.pagination.current_page,
page_size=result.pagination.page_size,
total_pages=result.pagination.total_pages,
)
@runs_router.get(
path="/{run_id}",
summary="Get run details",
response_model=RunDetails,
)
async def get_run(
run_id: str = Path(description="Run ID"),
auth: APIAuthorizationInfo = Security(
require_permission(APIKeyPermission.READ_RUN)
),
) -> RunDetails:
"""
Get detailed information about a specific run.
Includes outputs and individual node execution results.
"""
result = await execution_db.get_graph_execution(
user_id=auth.user_id,
execution_id=run_id,
include_node_executions=True,
)
if not result:
raise HTTPException(status_code=404, detail=f"Run #{run_id} not found")
return _convert_execution_to_run_details(result)
@runs_router.post(
path="/{run_id}/stop",
summary="Stop a run",
)
async def stop_run(
run_id: str = Path(description="Run ID"),
auth: APIAuthorizationInfo = Security(
require_permission(APIKeyPermission.WRITE_RUN)
),
) -> Run:
"""
Stop a running execution.
Only runs in QUEUED or RUNNING status can be stopped.
"""
# Verify the run exists and belongs to the user
exec = await execution_db.get_graph_execution(
user_id=auth.user_id,
execution_id=run_id,
)
if not exec:
raise HTTPException(status_code=404, detail=f"Run #{run_id} not found")
# Stop the execution
await execution_utils.stop_graph_execution(
graph_exec_id=run_id,
user_id=auth.user_id,
)
# Fetch updated execution
updated_exec = await execution_db.get_graph_execution(
user_id=auth.user_id,
execution_id=run_id,
)
if not updated_exec:
raise HTTPException(status_code=404, detail=f"Run #{run_id} not found")
return _convert_execution_to_run(updated_exec)
@runs_router.delete(
path="/{run_id}",
summary="Delete a run",
)
async def delete_run(
run_id: str = Path(description="Run ID"),
auth: APIAuthorizationInfo = Security(
require_permission(APIKeyPermission.WRITE_RUN)
),
) -> None:
"""
Delete an execution run.
This marks the run as deleted. The data may still be retained for
some time for recovery purposes.
"""
await execution_db.delete_graph_execution(
graph_exec_id=run_id,
user_id=auth.user_id,
)
# ============================================================================
# Endpoints - Reviews (Human-in-the-loop)
# ============================================================================
@runs_router.get(
path="/reviews",
summary="List all pending reviews",
response_model=PendingReviewsResponse,
)
async def list_pending_reviews(
auth: APIAuthorizationInfo = Security(
require_permission(APIKeyPermission.READ_RUN_REVIEW)
),
page: int = Query(default=1, ge=1, description="Page number (1-indexed)"),
page_size: int = Query(
default=DEFAULT_PAGE_SIZE,
ge=1,
le=MAX_PAGE_SIZE,
description=f"Items per page (max {MAX_PAGE_SIZE})",
),
) -> PendingReviewsResponse:
"""
List all pending human-in-the-loop reviews.
These are blocks that require human approval or input before the
agent can continue execution.
"""
reviews = await review_db.get_pending_reviews_for_user(
user_id=auth.user_id,
page=page,
page_size=page_size,
)
# Note: get_pending_reviews_for_user returns list directly, not a paginated result
# We compute pagination info based on results
total_count = len(reviews)
total_pages = max(1, (total_count + page_size - 1) // page_size)
return PendingReviewsResponse(
reviews=[_convert_pending_review(r) for r in reviews],
total_count=total_count,
page=page,
page_size=page_size,
total_pages=total_pages,
)
@runs_router.get(
path="/{run_id}/reviews",
summary="List reviews for a run",
response_model=list[PendingReview],
)
async def list_run_reviews(
run_id: str = Path(description="Run ID"),
auth: APIAuthorizationInfo = Security(
require_permission(APIKeyPermission.READ_RUN_REVIEW)
),
) -> list[PendingReview]:
"""
List all human-in-the-loop reviews for a specific run.
"""
reviews = await review_db.get_pending_reviews_for_execution(
graph_exec_id=run_id,
user_id=auth.user_id,
)
return [_convert_pending_review(r) for r in reviews]
@runs_router.post(
path="/{run_id}/reviews",
summary="Submit review responses for a run",
response_model=SubmitReviewsResponse,
)
async def submit_reviews(
request: SubmitReviewsRequest,
run_id: str = Path(description="Run ID"),
auth: APIAuthorizationInfo = Security(
require_permission(APIKeyPermission.WRITE_RUN_REVIEW)
),
) -> SubmitReviewsResponse:
"""
Submit responses to all pending human-in-the-loop reviews for a run.
All pending reviews for the execution must be included. Approving
a review will allow the agent to continue; rejecting will terminate
execution at that point.
"""
# Build review decisions dict for process_all_reviews_for_execution
review_decisions: dict[
str, tuple[ReviewStatus, SafeJsonData | None, str | None]
] = {}
for decision in request.reviews:
status = ReviewStatus.APPROVED if decision.approved else ReviewStatus.REJECTED
review_decisions[decision.node_exec_id] = (
status,
decision.edited_payload,
decision.message,
)
try:
results = await review_db.process_all_reviews_for_execution(
user_id=auth.user_id,
review_decisions=review_decisions,
)
approved_count = sum(
1 for r in results.values() if r.status == ReviewStatus.APPROVED
)
rejected_count = sum(
1 for r in results.values() if r.status == ReviewStatus.REJECTED
)
return SubmitReviewsResponse(
run_id=run_id,
approved_count=approved_count,
rejected_count=rejected_count,
)
except ValueError as e:
raise HTTPException(status_code=400, detail=str(e))

View File

@@ -0,0 +1,250 @@
"""
V2 External API - Schedules Endpoints
Provides endpoints for managing execution schedules.
"""
import logging
from datetime import datetime
from typing import Any, Optional
from fastapi import APIRouter, HTTPException, Path, Query, Security
from prisma.enums import APIKeyPermission
from pydantic import BaseModel, Field
from backend.api.external.middleware import require_permission
from backend.data import graph as graph_db
from backend.data.auth.base import APIAuthorizationInfo
from backend.data.user import get_user_by_id
from backend.executor import scheduler
from backend.util.clients import get_scheduler_client
from backend.util.timezone_utils import get_user_timezone_or_utc
from .common import DEFAULT_PAGE_SIZE, MAX_PAGE_SIZE
logger = logging.getLogger(__name__)
schedules_router = APIRouter()
# ============================================================================
# Request/Response Models
# ============================================================================
class Schedule(BaseModel):
"""An execution schedule for a graph."""
id: str
name: str
graph_id: str
graph_version: int
cron: str = Field(description="Cron expression for the schedule")
input_data: dict[str, Any] = Field(
default_factory=dict, description="Input data for scheduled executions"
)
next_run_time: Optional[datetime] = Field(
default=None, description="Next scheduled run time"
)
is_enabled: bool = Field(default=True, description="Whether schedule is enabled")
class SchedulesListResponse(BaseModel):
"""Response for listing schedules."""
schedules: list[Schedule]
total_count: int
page: int
page_size: int
total_pages: int
class CreateScheduleRequest(BaseModel):
"""Request to create a schedule."""
name: str = Field(description="Display name for the schedule")
cron: str = Field(description="Cron expression (e.g., '0 9 * * *' for 9am daily)")
input_data: dict[str, Any] = Field(
default_factory=dict, description="Input data for scheduled executions"
)
credentials_inputs: dict[str, Any] = Field(
default_factory=dict, description="Credentials for the schedule"
)
graph_version: Optional[int] = Field(
default=None, description="Graph version (default: active version)"
)
timezone: Optional[str] = Field(
default=None,
description=(
"Timezone for schedule (e.g., 'America/New_York'). "
"Defaults to user's timezone."
),
)
def _convert_schedule(job: scheduler.GraphExecutionJobInfo) -> Schedule:
"""Convert internal schedule job info to v2 API model."""
# Parse the ISO format string to datetime
next_run = datetime.fromisoformat(job.next_run_time) if job.next_run_time else None
return Schedule(
id=job.id,
name=job.name or "",
graph_id=job.graph_id,
graph_version=job.graph_version,
cron=job.cron,
input_data=job.input_data,
next_run_time=next_run,
is_enabled=True, # All returned schedules are enabled
)
# ============================================================================
# Endpoints
# ============================================================================
@schedules_router.get(
path="",
summary="List all user schedules",
response_model=SchedulesListResponse,
)
async def list_all_schedules(
auth: APIAuthorizationInfo = Security(
require_permission(APIKeyPermission.READ_SCHEDULE)
),
page: int = Query(default=1, ge=1, description="Page number (1-indexed)"),
page_size: int = Query(
default=DEFAULT_PAGE_SIZE,
ge=1,
le=MAX_PAGE_SIZE,
description=f"Items per page (max {MAX_PAGE_SIZE})",
),
) -> SchedulesListResponse:
"""
List all schedules for the authenticated user across all graphs.
"""
schedules = await get_scheduler_client().get_execution_schedules(
user_id=auth.user_id
)
converted = [_convert_schedule(s) for s in schedules]
# Manual pagination (scheduler doesn't support pagination natively)
total_count = len(converted)
total_pages = (total_count + page_size - 1) // page_size if total_count > 0 else 1
start = (page - 1) * page_size
end = start + page_size
paginated = converted[start:end]
return SchedulesListResponse(
schedules=paginated,
total_count=total_count,
page=page,
page_size=page_size,
total_pages=total_pages,
)
@schedules_router.delete(
path="/{schedule_id}",
summary="Delete a schedule",
)
async def delete_schedule(
schedule_id: str = Path(description="Schedule ID to delete"),
auth: APIAuthorizationInfo = Security(
require_permission(APIKeyPermission.WRITE_SCHEDULE)
),
) -> None:
"""
Delete an execution schedule.
"""
try:
await get_scheduler_client().delete_schedule(
schedule_id=schedule_id,
user_id=auth.user_id,
)
except Exception as e:
if "not found" in str(e).lower():
raise HTTPException(
status_code=404, detail=f"Schedule #{schedule_id} not found"
)
raise
# ============================================================================
# Graph-specific Schedule Endpoints (nested under /graphs)
# These are included in the graphs router via include_router
# ============================================================================
graph_schedules_router = APIRouter()
@graph_schedules_router.get(
path="/{graph_id}/schedules",
summary="List schedules for a graph",
response_model=list[Schedule],
)
async def list_graph_schedules(
graph_id: str = Path(description="Graph ID"),
auth: APIAuthorizationInfo = Security(
require_permission(APIKeyPermission.READ_SCHEDULE)
),
) -> list[Schedule]:
"""
List all schedules for a specific graph.
"""
schedules = await get_scheduler_client().get_execution_schedules(
user_id=auth.user_id,
graph_id=graph_id,
)
return [_convert_schedule(s) for s in schedules]
@graph_schedules_router.post(
path="/{graph_id}/schedules",
summary="Create a schedule for a graph",
response_model=Schedule,
)
async def create_graph_schedule(
request: CreateScheduleRequest,
graph_id: str = Path(description="Graph ID"),
auth: APIAuthorizationInfo = Security(
require_permission(APIKeyPermission.WRITE_SCHEDULE)
),
) -> Schedule:
"""
Create a new execution schedule for a graph.
The schedule will execute the graph at times matching the cron expression,
using the provided input data.
"""
graph = await graph_db.get_graph(
graph_id=graph_id,
version=request.graph_version,
user_id=auth.user_id,
)
if not graph:
raise HTTPException(
status_code=404,
detail=f"Graph #{graph_id} v{request.graph_version} not found.",
)
# Determine timezone
if request.timezone:
user_timezone = request.timezone
else:
user = await get_user_by_id(auth.user_id)
user_timezone = get_user_timezone_or_utc(user.timezone if user else None)
result = await get_scheduler_client().add_execution_schedule(
user_id=auth.user_id,
graph_id=graph_id,
graph_version=graph.version,
name=request.name,
cron=request.cron,
input_data=request.input_data,
input_credentials=request.credentials_inputs,
user_timezone=user_timezone,
)
return _convert_schedule(result)

View File

@@ -693,13 +693,13 @@ class DeleteGraphResponse(TypedDict):
async def list_graphs(
user_id: Annotated[str, Security(get_user_id)],
) -> Sequence[graph_db.GraphMeta]:
paginated_result = await graph_db.list_graphs_paginated(
graphs, _ = await graph_db.list_graphs_paginated(
user_id=user_id,
page=1,
page_size=250,
filter_by="active",
)
return paginated_result.graphs
return graphs
@v1_router.get(

View File

@@ -804,9 +804,7 @@ class GraphModel(Graph):
)
class GraphMeta(Graph):
user_id: str
class GraphMeta(GraphModel):
# Easy work-around to prevent exposing nodes and links in the API response
nodes: list[NodeModel] = Field(default=[], exclude=True) # type: ignore
links: list[Link] = Field(default=[], exclude=True)
@@ -816,13 +814,6 @@ class GraphMeta(Graph):
return GraphMeta(**graph.model_dump())
class GraphsPaginated(BaseModel):
"""Response schema for paginated graphs."""
graphs: list[GraphMeta]
pagination: Pagination
# --------------------- CRUD functions --------------------- #
@@ -856,7 +847,7 @@ async def list_graphs_paginated(
page: int = 1,
page_size: int = 25,
filter_by: Literal["active"] | None = "active",
) -> GraphsPaginated:
) -> tuple[list[GraphMeta], Pagination]:
"""
Retrieves paginated graph metadata objects.
@@ -867,7 +858,8 @@ async def list_graphs_paginated(
filter_by: An optional filter to either select graphs.
Returns:
GraphsPaginated: Paginated list of graph metadata.
list[GraphMeta]: List of graph info objects.
Pagination: Pagination information.
"""
where_clause: AgentGraphWhereInput = {"userId": user_id}
@@ -900,14 +892,11 @@ async def list_graphs_paginated(
logger.error(f"Error processing graph {graph.id}: {e}")
continue
return GraphsPaginated(
graphs=graph_models,
pagination=Pagination(
total_items=total_count,
total_pages=total_pages,
current_page=page,
page_size=page_size,
),
return graph_models, Pagination(
total_items=total_count,
total_pages=total_pages,
current_page=page,
page_size=page_size,
)

View File

@@ -1040,16 +1040,31 @@ enum SubmissionStatus {
}
enum APIKeyPermission {
// Legacy v1 permissions (kept for backward compatibility)
IDENTITY // Info about the authenticated user
EXECUTE_GRAPH // Can execute agent graphs
EXECUTE_GRAPH // Can execute agent graphs (v1 only)
READ_GRAPH // Can get graph versions and details
EXECUTE_BLOCK // Can execute individual blocks
EXECUTE_BLOCK // Can execute individual blocks (v1 only)
READ_BLOCK // Can get block information
READ_STORE // Can read store agents and creators
USE_TOOLS // Can use chat tools via external API
MANAGE_INTEGRATIONS // Can initiate OAuth flows and complete them
READ_STORE // Can read store/marketplace agents and creators
USE_TOOLS // Can use chat tools via external API (v1 only)
MANAGE_INTEGRATIONS // Can initiate OAuth flows and complete them (v1 only)
READ_INTEGRATIONS // Can list credentials and providers
DELETE_INTEGRATIONS // Can delete credentials
DELETE_INTEGRATIONS // Can delete credentials (v1 only)
// V2 permissions
WRITE_GRAPH // Can create, update, delete graphs
READ_SCHEDULE // Can list schedules
WRITE_SCHEDULE // Can create and delete schedules
WRITE_STORE // Can create, update, delete marketplace submissions
READ_LIBRARY // Can list library agents and runs
RUN_AGENT // Can run agents from library
READ_RUN // Can list and get run details
WRITE_RUN // Can stop and delete runs
READ_RUN_REVIEW // Can list pending human-in-the-loop reviews
WRITE_RUN_REVIEW // Can submit human-in-the-loop review responses
READ_CREDITS // Can get credit balance and transactions
UPLOAD_FILES // Can upload files for agent input
}
model APIKey {

View File

@@ -2,7 +2,7 @@
## Overview
The AutoGPT platform implements OAuth 2.0 in two distinct contexts:
The AutoGPT Platform implements OAuth 2.0 in two distinct contexts:
1. **User Authentication (SSO)**: Handled by Supabase for platform login
2. **API Integration Credentials**: Custom OAuth implementation for third-party service access
@@ -324,7 +324,7 @@ stateDiagram-v2
### User Authentication (SSO) via Supabase
- **Purpose**: Authenticate users to access the AutoGPT platform
- **Purpose**: Authenticate users to access the AutoGPT Platform
- **Provider**: Supabase Auth (currently supports Google SSO)
- **Flow Path**: `/login` → Supabase OAuth → `/auth/callback`
- **Session Storage**: Supabase-managed cookies

View File

@@ -1,14 +1,18 @@
# AutoGPT Platform External API Guide
The AutoGPT Platform provides an External API that allows you to programmatically interact with agents, blocks, the store, and more.
The AutoGPT Platform provides an External API that allows you to programmatically interact with agents, blocks, the marketplace, and more.
## API Documentation
Full API documentation with interactive examples is available at:
**[https://backend.agpt.co/external-api/docs](https://backend.agpt.co/external-api/docs)**
- **Main**: [https://backend.agpt.co/external-api/docs](https://backend.agpt.co/external-api/docs)
- **v2 API**: [https://backend.agpt.co/external-api/v2/docs](https://backend.agpt.co/external-api/v2/docs)
- **v1 API**: [https://backend.agpt.co/external-api/v1/docs](https://backend.agpt.co/external-api/v1/docs)
This Swagger UI documentation includes all available endpoints, request/response schemas, and allows you to try out API calls directly.
The Swagger UI documentation includes all available endpoints, request/response schemas, and allows you to try out API calls directly.
**Recommendation**: New integrations should use the v2 API.
## Authentication Methods
@@ -16,11 +20,12 @@ The External API supports two authentication methods:
### 1. API Keys
API keys are the simplest way to authenticate. Generate an API key from your AutoGPT Platform account settings and include it in your requests:
API keys are the simplest way to authenticate. Generate an API key from your AutoGPT Platform account settings and include it in your requests using the `X-API-Key` header:
```http
GET /external-api/v1/blocks
X-API-Key: your_api_key_here
```bash
# List available blocks
curl -H "X-API-Key: YOUR_API_KEY" \
https://backend.agpt.co/external-api/v1/blocks
```
API keys are ideal for:
@@ -32,51 +37,62 @@ API keys are ideal for:
For applications that need to act on behalf of users, use OAuth 2.0. This allows users to authorize your application to access their AutoGPT resources.
OAuth is ideal for:
- Third-party applications
- "Sign in with AutoGPT" (SSO, Single Sign-On) functionality
- Applications that need user-specific permissions
See the [SSO Integration Guide](sso-guide.md) for complete OAuth implementation details.
## Available Scopes
When using OAuth, request only the scopes your application needs:
| Scope | Description |
|-------|-------------|
| `IDENTITY` | Read user ID, e-mail, and timezone |
| `EXECUTE_GRAPH` | Run agents |
| `READ_GRAPH` | Read agent run results |
| `EXECUTE_BLOCK` | Run individual blocks |
| `READ_BLOCK` | Read block definitions |
| `READ_STORE` | Access the agent store |
| `USE_TOOLS` | Use platform tools |
| `MANAGE_INTEGRATIONS` | Create and update user integrations |
| `READ_INTEGRATIONS` | Read user integration status |
| `DELETE_INTEGRATIONS` | Remove user integrations |
## Quick Start
### Using an API Key
```bash
# List available blocks
curl -H "X-API-Key: YOUR_API_KEY" \
https://backend.agpt.co/external-api/v1/blocks
```
### Using OAuth
To get started:
1. Register an OAuth application (contact platform administrator)
2. Implement the OAuth flow as described in the [SSO Guide](sso-guide.md)
3. Use the obtained access token:
2. Implement the OAuth flow as described in the [OAuth Guide](oauth-guide.md)
3. Go through the OAuth flow to authorize your app and obtain an access token
4. Make API requests with the access token in the `Authorization: Bearer` header:
```bash
curl -H "Authorization: Bearer agpt_xt_..." \
https://backend.agpt.co/external-api/v1/blocks
```
OAuth is ideal for:
- Third-party applications
- "Sign in with AutoGPT" (SSO, Single Sign-On) functionality
- Applications that need user-specific permissions
See the [OAuth Integration Guide](oauth-guide.md) for complete OAuth implementation details.
## Available Scopes
When creating API keys or using OAuth, request only the scopes your application needs.
### Core Scopes
| Scope | Description |
|-------|-------------|
| `IDENTITY` | Read user ID, e-mail, and timezone |
| `READ_GRAPH` | Read graph/agent definitions and versions |
| `WRITE_GRAPH` | Create, update, and delete graphs |
| `READ_BLOCK` | Read block definitions |
| `READ_STORE` | Access the agent marketplace |
| `WRITE_STORE` | Create, update, and delete marketplace submissions |
| `READ_LIBRARY` | List library agents and their runs |
| `RUN_AGENT` | Execute agents from your library |
| `READ_RUN` | List and get execution run details |
| `WRITE_RUN` | Stop and delete runs |
| `READ_RUN_REVIEW` | List pending human-in-the-loop reviews |
| `WRITE_RUN_REVIEW` | Submit human-in-the-loop review responses |
| `READ_SCHEDULE` | List execution schedules |
| `WRITE_SCHEDULE` | Create and delete schedules |
| `READ_CREDITS` | Get credit balance and transaction history |
| `READ_INTEGRATIONS` | List OAuth credentials |
| `UPLOAD_FILES` | Upload files for agent input |
### Legacy Scopes (v1 only)
| Scope | Description |
|-------|-------------|
| `EXECUTE_GRAPH` | Execute graphs directly (use `RUN_AGENT` in v2) |
| `EXECUTE_BLOCK` | Execute individual blocks |
| `USE_TOOLS` | Use chat tools via external API |
| `MANAGE_INTEGRATIONS` | Initiate and complete OAuth flows |
| `DELETE_INTEGRATIONS` | Delete OAuth credentials |
## Support
For issues or questions about API integration: