Skip to main content

Adding a New Integration

Integration Scaffolding

4 files. One pattern. Auto-generated everything.

Write adapter.py + service.py — routes, CLI commands, and MCP tools are generated automatically.

HTTP Client
adapter.py
httpx calls
Bearer token injection
raise_for_status()
called by
←
Service Layer
service.py
@service_method
Clean method signatures
Param validation
the only file you reason about at call time
auto-generates
→
Router
routes.py
build_router(ServiceService)
CLI Commands
constellation mynewservice <method>
MCP Tools
Registered in /v1/meta
OAuth Flow (if needed)
oauth.py
Subclass OAuthHandler · Declares authorize_url, token_url, scopes
Handles callback independently — does not affect service.py

What & Why​

Every integration in Octopus is 4 files in a fixed pattern. The pattern is rigid by design — it ensures every integration gets HTTP routes, CLI commands, and MCP tool registrations for free, without any per-integration boilerplate.

FileYour jobWhat it produces
adapter.pyWrite all raw httpx calls to the external APIIsolated HTTP client, auth injection
service.pyExpose clean methods decorated with @service_methodCalled by routes, CLI, and MCP tools
routes.pyOne-liner: call build_router(ServiceService)FastAPI router mounted under /v1/{service}/
oauth.pySubclass OAuthHandler, declare OAuth configOAuth authorize + callback endpoints

The adapter is the only file that knows about the external API's specifics. The service is the only file that consumers interact with. Everything else is infrastructure.

Step-by-Step Guide​

Step 1 — Create the integration directory​

mkdir -p apps/constellation/app/integrations/mynewservice
touch apps/constellation/app/integrations/mynewservice/__init__.py

Step 2 — Create adapter.py​

# apps/constellation/app/integrations/mynewservice/adapter.py

from app.core.context import RequestContext
from app.core.http import get_http_client


class MyNewServiceAdapter:
    def __init__(self, context: RequestContext):
        self.context = context
        self.base_url = "https://api.mynewservice.com/v1"
        self.client = get_http_client()

    async def _get_headers(self) -> dict:
        return {"Authorization": f"Bearer {self.context.bearer_token}"}

    async def get_resource(self, resource_id: str) -> dict:
        headers = await self._get_headers()
        response = await self.client.get(
            f"{self.base_url}/resources/{resource_id}",
            headers=headers,
        )
        response.raise_for_status()
        return response.json()

    async def list_resources(self, limit: int = 20, offset: int = 0) -> dict:
        headers = await self._get_headers()
        response = await self.client.get(
            f"{self.base_url}/resources",
            headers=headers,
            params={"limit": limit, "offset": offset},
        )
        response.raise_for_status()
        return response.json()

    async def create_resource(self, name: str, description: str = "") -> dict:
        headers = await self._get_headers()
        response = await self.client.post(
            f"{self.base_url}/resources",
            headers=headers,
            json={"name": name, "description": description},
        )
        response.raise_for_status()
        return response.json()

    async def update_resource(self, resource_id: str, name: str = None, description: str = None) -> dict:
        headers = await self._get_headers()
        payload = {k: v for k, v in {"name": name, "description": description}.items() if v is not None}
        response = await self.client.patch(
            f"{self.base_url}/resources/{resource_id}",
            headers=headers,
            json=payload,
        )
        response.raise_for_status()
        return response.json()

    async def delete_resource(self, resource_id: str) -> dict:
        headers = await self._get_headers()
        response = await self.client.delete(
            f"{self.base_url}/resources/{resource_id}",
            headers=headers,
        )
        response.raise_for_status()
        return {"deleted": True, "resource_id": resource_id}

Rules for adapter.py:

  • Accept only primitive types (str, int, bool, dict, list) — no Pydantic models.
  • Always call response.raise_for_status() before returning.
  • Use self.context.bearer_token for auth. For API key auth, use self.context.api_key.
  • One adapter method per API endpoint. No business logic here.

Step 3 — Create service.py​

# apps/constellation/app/integrations/mynewservice/service.py

from app.core.context import RequestContext
from app.core.decorators import service_method
from .adapter import MyNewServiceAdapter


class MyNewServiceService:
    def __init__(self, context: RequestContext):
        self.adapter = MyNewServiceAdapter(context)

    @service_method(verb="GET", tags=["mynewservice"])
    async def get_resource(self, resource_id: str) -> dict:
        """Get a resource by ID."""
        return await self.adapter.get_resource(resource_id)

    @service_method(verb="GET", tags=["mynewservice"])
    async def list_resources(self, limit: int = 20, offset: int = 0) -> dict:
        """List resources with optional pagination."""
        return await self.adapter.list_resources(limit=limit, offset=offset)

    @service_method(verb="POST", tags=["mynewservice"])
    async def create_resource(self, name: str, description: str = "") -> dict:
        """Create a new resource."""
        return await self.adapter.create_resource(name=name, description=description)

    @service_method(verb="PATCH", tags=["mynewservice"], path_params=["resource_id"])
    async def update_resource(
        self,
        resource_id: str,
        name: str = None,
        description: str = None,
    ) -> dict:
        """Update an existing resource."""
        return await self.adapter.update_resource(
            resource_id=resource_id,
            name=name,
            description=description,
        )

    @service_method(verb="DELETE", tags=["mynewservice"], path_params=["resource_id"])
    async def delete_resource(self, resource_id: str) -> dict:
        """Delete a resource by ID."""
        return await self.adapter.delete_resource(resource_id)

Rules for service.py:

  • Every public method must have @service_method.
  • The docstring becomes the description in /v1/meta, CLI --help, and MCP tool descriptions. Write it clearly.
  • Parameters listed in path_params are appended to the URL path (/v1/mynewservice/update_resource/{resource_id}); all other parameters go in the request body.

Step 4 — Create routes.py​

# apps/constellation/app/integrations/mynewservice/routes.py

from app.core.generator import build_router
from .service import MyNewServiceService

router = build_router(MyNewServiceService)

This is the complete file. build_router inspects MyNewServiceService, reads every @service_method decorator, and generates the FastAPI routes automatically. Do not add anything else here.

Step 5 — Create oauth.py (OAuth integrations only)​

Skip this step if the integration uses API key authentication.

# apps/constellation/app/integrations/mynewservice/oauth.py

from app.core.auth import OAuthHandler


class MyNewServiceOAuthHandler(OAuthHandler):
    service_name = "mynewservice"
    authorize_url = "https://mynewservice.com/oauth/authorize"
    token_url = "https://mynewservice.com/oauth/token"
    scopes = ["read", "write"]
    client_id_env = "MYNEWSERVICE_CLIENT_ID"
    client_secret_env = "MYNEWSERVICE_CLIENT_SECRET"

This registers two endpoints automatically:

  • POST /v1/oauth/mynewservice/authorize — returns { "authorization_url": "..." }
  • GET /v1/oauth/mynewservice/callback — exchanges the code for a token

Add the required environment variables to .env.example:

# MyNewService OAuth credentials
MYNEWSERVICE_CLIENT_ID=
MYNEWSERVICE_CLIENT_SECRET=

Step 6 — Import in main.py​

Open apps/constellation/app/main.py and add an import for the new integration module. The @service_method decorator auto-registers routes when the module is imported — no manual app.include_router() call is needed.

# apps/constellation/app/main.py  (add this import alongside existing integrations)

import app.integrations.mynewservice  # noqa: F401

If the integration has an oauth.py, also register the OAuth handler:

from app.integrations.mynewservice.oauth import MyNewServiceOAuthHandler

oauth_registry.register(MyNewServiceOAuthHandler)

Step 7 — Test via HTTP​

Start the server and test the new endpoints:

# List resources
curl -X POST http://localhost:8000/v1/mynewservice/list_resources \
  -H "Authorization: Bearer $TOKEN" \
  -H "Content-Type: application/json" \
  -d '{"limit": 5}'

# Get a specific resource
curl -X POST http://localhost:8000/v1/mynewservice/get_resource \
  -H "Authorization: Bearer $TOKEN" \
  -H "Content-Type: application/json" \
  -d '{"resource_id": "abc123"}'

# Create a resource
curl -X POST http://localhost:8000/v1/mynewservice/create_resource \
  -H "Authorization: Bearer $TOKEN" \
  -H "Content-Type: application/json" \
  -d '{"name": "My Resource", "description": "Created via Octopus"}'

# Delete a resource (resource_id is a path param)
curl -X DELETE http://localhost:8000/v1/mynewservice/delete_resource/abc123 \
  -H "Authorization: Bearer $TOKEN"

Step 8 — Verify CLI commands​

The CLI is auto-generated from the same service methods:

# List all commands for the new integration
constellation mynewservice --help

# Run a method
constellation mynewservice list-resources --limit 5
constellation mynewservice get-resource --resource-id abc123
constellation mynewservice create-resource --name "My Resource" --description "Test"
constellation mynewservice delete-resource --resource-id abc123

Method names use snake_case in Python and are exposed as kebab-case in the CLI (get_resource → get-resource, list_resources → list-resources).

Step 9 — Verify in /v1/meta​

All registered methods appear in the meta schema. Confirm your new integration is present:

curl http://localhost:8000/v1/meta | jq '.services.mynewservice'

Expected response shape:

{
  "get_resource": {
    "verb": "GET",
    "description": "Get a resource by ID.",
    "tags": ["mynewservice"],
    "parameters": {
      "resource_id": { "type": "string", "required": true }
    }
  },
  "list_resources": { "..." : "..." },
  "create_resource": { "..." : "..." },
  "update_resource": { "..." : "..." },
  "delete_resource": { "..." : "..." }
}
@service_method Parameters

All decorator options and what they control

Every parameter affects how routes, CLI commands, and MCP tools are generated.

verb
HTTP method for the generated route.
"GET"
"POST"
"PUT"
"PATCH"
"DELETE"
Default: "POST". GET requests pass all params as query strings.
tags
List of strings for grouping in the OpenAPI schema and CLI.
tags=["mynewservice"]
tags=["mynewservice", "admin"]
Multiple tags make the method appear in multiple groups in the CLI help output.
path_params
Parameter names that are appended to the URL path instead of the request body.
path_params=["resource_id"]
# generates route:
/v1/svc/update_resource/{resource_id}
Use for resource IDs in PUT, PATCH, and DELETE methods. Omit for POST body params.
description
Override the method docstring for the generated schema and CLI help.
description="Fetch a resource by its unique ID."
Optional. When omitted, the Python docstring is used. Keep docstrings clear — they become MCP tool descriptions.
@service_method(verb="PATCH", tags=["mynewservice"], path_params=["resource_id"], description="Update a resource.")

Completion Checklist​

Use this checklist to confirm your integration is complete before committing.

Integration Checklist