Skip to content

Artifacts, projects, MCP and the rest

octomate.schemas.artifacts

Owned artifacts; a thread may reference one but does not own its content.

ArtifactSandboxPolicy pydantic-model

Bases: BaseModel

Public sandbox policy, tied to the embedding console by frame-ancestors.

Sources covering the console hostname are refused, including wildcards and alternate schemes or ports. HTML and invocation data never travel in this policy URL; the authenticated viewer sends them over the sandbox bridge.

Fields:

Validators:

Source code in octomate/schemas/artifacts.py 
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
class ArtifactSandboxPolicy(BaseModel):
    """Public sandbox policy, tied to the embedding console by frame-ancestors.

    Sources covering the console hostname are refused, including wildcards and
    alternate schemes or ports. HTML and invocation data never travel in this
    policy URL; the authenticated viewer sends them over the sandbox bridge.
    """

    parent_origin: AnyHttpUrl = Field(description="The embedding console's origin.")
    csp: McpAppResourceCSP = Field(
        default_factory=McpAppResourceCSP, description="Validated resource allowances."
    )

    @model_validator(mode="after")
    def protect_console(self) -> "ArtifactSandboxPolicy":
        parent = self.parent_origin
        if (
            parent.username is not None
            or parent.password is not None
            or parent.path not in {None, "/"}
            or parent.query is not None
            or parent.fragment is not None
            or parent.host is None
        ):
            raise ValueError("The sandbox parent must be an origin")
        for sources in (
            self.csp.connect_domains,
            self.csp.resource_domains,
            self.csp.frame_domains,
            self.csp.base_uri_domains,
        ):
            for source in sources or []:
                host = AnyUrl(source).host
                if host == parent.host or (
                    host is not None
                    and host.startswith("*.")
                    and parent.host.endswith(host[1:])
                ):
                    raise ValueError("Artifact policy must not allow the console host")
        return self

parent_origin pydantic-field 

parent_origin: AnyHttpUrl

The embedding console's origin.

csp pydantic-field 

csp: McpAppResourceCSP

Validated resource allowances.

protect_console pydantic-validator 

protect_console() -> ArtifactSandboxPolicy
Source code in octomate/schemas/artifacts.py 
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
@model_validator(mode="after")
def protect_console(self) -> "ArtifactSandboxPolicy":
    parent = self.parent_origin
    if (
        parent.username is not None
        or parent.password is not None
        or parent.path not in {None, "/"}
        or parent.query is not None
        or parent.fragment is not None
        or parent.host is None
    ):
        raise ValueError("The sandbox parent must be an origin")
    for sources in (
        self.csp.connect_domains,
        self.csp.resource_domains,
        self.csp.frame_domains,
        self.csp.base_uri_domains,
    ):
        for source in sources or []:
            host = AnyUrl(source).host
            if host == parent.host or (
                host is not None
                and host.startswith("*.")
                and parent.host.endswith(host[1:])
            ):
                raise ValueError("Artifact policy must not allow the console host")
    return self

Artifact

Bases: BaseTransmuter

A deliverable with its own identity and owner, requiring no thread.

The artifact UUID identifies its viewer independently of any ledger row. Referencing it in a thread does not transfer ownership or grant content access.

Source code in octomate/schemas/artifacts.py 
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
@sqlalchemy_materia.bless(artifact_models.Artifact)
class Artifact(BaseTransmuter):
    """A deliverable with its own identity and owner, requiring no thread.

    The artifact UUID identifies its viewer independently of any ledger row.
    Referencing it in a thread does not transfer ownership or grant content access.
    """

    model_config = ConfigDict(from_attributes=True)

    id: Annotated[UUID7, Identity] = Field(default_factory=uuid7, frozen=True)
    owner_id: UUID7 = Field(
        frozen=True,
        description="Registered owner; thread participation does not replace this ownership.",
    )
    title: str = Field(
        min_length=1, description="Display title captured when the artifact is created."
    )
    created_at: AwareDatetime = Field(
        default_factory=lambda: datetime.now(UTC),
        description="When this artifact was recorded, in UTC.",
    )

model_config class-attribute instance-attribute 

model_config = ConfigDict(from_attributes=True)

id class-attribute instance-attribute 

id: Annotated[UUID7, Identity] = Field(
    default_factory=uuid7, frozen=True
)

owner_id class-attribute instance-attribute 

owner_id: UUID7 = Field(
    frozen=True,
    description="Registered owner; thread participation does not replace this ownership.",
)

title class-attribute instance-attribute 

title: str = Field(
    min_length=1,
    description="Display title captured when the artifact is created.",
)

created_at class-attribute instance-attribute 

created_at: AwareDatetime = Field(
    default_factory=lambda: datetime.now(UTC),
    description="When this artifact was recorded, in UTC.",
)

McpAppArtifact

Bases: Artifact

An MCP invocation retained for an authorized artifact viewer.

The saved input and result support replay after an authorized UI resource fetch. The invocation's installation UUID never follows namespace reuse, and deleting a thread or uninstalling the provider does not delete it.

Source code in octomate/schemas/artifacts.py 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
@sqlalchemy_materia.bless(artifact_models.McpAppArtifact)
class McpAppArtifact(Artifact):
    """An MCP invocation retained for an authorized artifact viewer.

    The saved input and result support replay after an authorized UI resource
    fetch. The invocation's installation UUID never follows namespace
    reuse, and deleting a thread or uninstalling the provider does not delete it.
    """

    kind: Literal["mcp_app"] = Field(default="mcp_app", frozen=True)
    invocation: McpAppInvocation = Field(
        repr=False,
        description="Original call and complete result, kept outside generic ledger responses.",
    )

kind class-attribute instance-attribute 

kind: Literal["mcp_app"] = Field(
    default="mcp_app", frozen=True
)

invocation class-attribute instance-attribute 

invocation: McpAppInvocation = Field(
    repr=False,
    description="Original call and complete result, kept outside generic ledger responses.",
)

McpAppOverview pydantic-model

Bases: BaseModel

Saved display metadata for native channel cards, without invocation content.

Fields:

Source code in octomate/schemas/artifacts.py 
107
108
109
110
111
112
113
114
115
116
117
class McpAppOverview(BaseModel):
    """Saved display metadata for native channel cards, without invocation content."""

    artifact_id: UUID7 = Field(description="The artifact's independent identity.")
    title: str = Field(description="Display title saved when the artifact was created.")
    provider_name: str = Field(
        description="Installation display name saved at execution."
    )
    viewer_url: AnyHttpUrl = Field(
        description="Absolute viewer link; opening it still requires the owner's login."
    )

artifact_id pydantic-field 

artifact_id: UUID7

The artifact's independent identity.

title pydantic-field 

title: str

Display title saved when the artifact was created.

provider_name pydantic-field 

provider_name: str

Installation display name saved at execution.

viewer_url pydantic-field 

viewer_url: AnyHttpUrl

Absolute viewer link; opening it still requires the owner's login.

McpAppView pydantic-model

Bases: BaseModel

An authorized artifact snapshot and freshly fetched resource for replay.

Fields:

Source code in octomate/schemas/artifacts.py 
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
class McpAppView(BaseModel):
    """An authorized artifact snapshot and freshly fetched resource for replay."""

    artifact: McpAppArtifact = Field(
        description="The artifact and complete saved invocation."
    )
    resource: McpAppResource = Field(
        description="Untrusted HTML and resource UI metadata."
    )
    sandbox_origin: AnyHttpUrl = Field(
        description="Configured origin for the isolated sandbox."
    )
    sandbox_url: AnyHttpUrl = Field(
        description="Sandbox proxy URL carrying the validated public policy only."
    )
    viewer_path: str = Field(
        description="Root-relative standalone artifact viewer URL."
    )

artifact pydantic-field 

artifact: McpAppArtifact

The artifact and complete saved invocation.

resource pydantic-field 

resource: McpAppResource

Untrusted HTML and resource UI metadata.

sandbox_origin pydantic-field 

sandbox_origin: AnyHttpUrl

Configured origin for the isolated sandbox.

sandbox_url pydantic-field 

sandbox_url: AnyHttpUrl

Sandbox proxy URL carrying the validated public policy only.

viewer_path pydantic-field 

viewer_path: str

Root-relative standalone artifact viewer URL.

octomate.schemas.project

LocalPath module-attribute 

LocalPath = Annotated[Path, BeforeValidator(local_path)]

UpstreamVariant 

UpstreamVariant = Annotated[
    RemoteUpstream | DirectoryUpstream,
    Field(discriminator="kind"),
]

RemoteUpstream pydantic-model

Bases: BaseModel

An upstream whose mirror is made by cloning and kept current by fetching.

Fields:

  • kind (Literal['remote'])
  • url (str)
Source code in octomate/schemas/project.py 
54
55
56
57
58
class RemoteUpstream(BaseModel):
    """An upstream whose mirror is made by cloning and kept current by fetching."""

    kind: Literal["remote"] = "remote"
    url: str = Field(description="The git remote the mirror clones and fetches from.")

kind pydantic-field 

kind: Literal['remote'] = 'remote'

url pydantic-field 

url: str

The git remote the mirror clones and fetches from.

DirectoryUpstream pydantic-model

Bases: BaseModel

An upstream whose mirror is made by git init and kept current by copying the directory in and committing.

Fields:

Source code in octomate/schemas/project.py 
61
62
63
64
65
66
67
68
class DirectoryUpstream(BaseModel):
    """An upstream whose mirror is made by `git init` and kept current by copying
    the directory in and committing."""

    kind: Literal["directory"] = "directory"
    path: LocalPath = Field(
        description="The local directory the mirror is synced from."
    )

kind pydantic-field 

kind: Literal['directory'] = 'directory'

path pydantic-field 

path: LocalPath

The local directory the mirror is synced from.

Project

Bases: BaseTransmuter

A project: the roots that are it, and how it is driven.

Every project is declared — the operator's projects: block is the only way in. A session running where no project claims is recorded under none.

Source code in octomate/schemas/project.py 
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
@sqlalchemy_materia.bless(ProjectModel)
class Project(BaseTransmuter):
    """A project: the roots that are it, and how it is driven.

    Every project is declared — the operator's ``projects:`` block is the only way in.
    A session running where no project claims is recorded under none.
    """

    model_config = ConfigDict(from_attributes=True)

    id: Annotated[UUID7, Identity] = Field(default_factory=uuid7, frozen=True)
    name: str = Field(
        default="",
        description=(
            "Stable name for this project. Defaults to the root's directory name — "
            "`~/Projects/octoverse/inky` is `inky` — and steps aside to `<name>-2` "
            "when another root already answers to it."
        ),
    )
    root: LocalPath = Field(
        description="The directory this project is, as an absolute local path."
    )
    extra_roots: list[LocalPath] = Field(
        default_factory=list,
        description=(
            "Further directories that are also this project — a settings tree, a "
            "sibling checkout — each an absolute local path."
        ),
    )
    upstream: UpstreamVariant = Field(
        description=(
            "Where this project's mirror comes from and how it is kept current: a "
            "remote is cloned and fetched, a directory is `git init`'d and synced by "
            "copying in and committing. Required: where a mirror syncs from is the "
            "declaration's to say, not guessed from the root."
        ),
    )
    description: str | None = Field(
        default=None,
        description=(
            "What this project is, for a human reading the registry and for a model "
            "choosing between registered projects. Not agent instructions: those live "
            "in the project's own `AGENTS.md`/`CLAUDE.md`."
        ),
    )
    enabled: bool = Field(
        default=True,
        description=(
            "Whether this project is part of the working set. Reconciliation clears "
            "it for a root that is no longer on disk, so the row survives for the "
            "runs and threads that name it while nothing new resolves to it."
        ),
    )

    @model_validator(mode="after")
    def name_from_root(self) -> Self:
        """A project with no name of its own is called after its root's directory.

        `local_path` has already refused a root with no name, so there is always one
        to take. A name two projects both want is refused by
        `ProjectManager.reconcile`, which is where the names in use are known.
        """
        self.name = self.name or self.root.name
        return self

    @property
    def roots(self) -> tuple[Path, ...]:
        """Every directory that is this project: `root`, then any `extra_roots`."""
        return (self.root, *self.extra_roots)

    def contains(self, path: Path) -> bool:
        """Whether `path` is one of this project's roots, or inside one.

        Both sides are resolved, because `is_relative_to` is lexical: a `..` and a
        symlink are the whole attack surface of a check that compares the string it
        was handed. A path that does not exist resolves fine, which is what a write
        creating a file needs.
        """
        resolved = path.resolve()
        return any(resolved.is_relative_to(root.resolve()) for root in self.roots)

model_config class-attribute instance-attribute 

model_config = ConfigDict(from_attributes=True)

id class-attribute instance-attribute 

id: Annotated[UUID7, Identity] = Field(
    default_factory=uuid7, frozen=True
)

name class-attribute instance-attribute 

name: str = Field(
    default="",
    description="Stable name for this project. Defaults to the root's directory name — `~/Projects/octoverse/inky` is `inky` — and steps aside to `<name>-2` when another root already answers to it.",
)

root class-attribute instance-attribute 

root: LocalPath = Field(
    description="The directory this project is, as an absolute local path."
)

extra_roots class-attribute instance-attribute 

extra_roots: list[LocalPath] = Field(
    default_factory=list,
    description="Further directories that are also this project — a settings tree, a sibling checkout — each an absolute local path.",
)

upstream class-attribute instance-attribute 

upstream: UpstreamVariant = Field(
    description="Where this project's mirror comes from and how it is kept current: a remote is cloned and fetched, a directory is `git init`'d and synced by copying in and committing. Required: where a mirror syncs from is the declaration's to say, not guessed from the root."
)

description class-attribute instance-attribute 

description: str | None = Field(
    default=None,
    description="What this project is, for a human reading the registry and for a model choosing between registered projects. Not agent instructions: those live in the project's own `AGENTS.md`/`CLAUDE.md`.",
)

enabled class-attribute instance-attribute 

enabled: bool = Field(
    default=True,
    description="Whether this project is part of the working set. Reconciliation clears it for a root that is no longer on disk, so the row survives for the runs and threads that name it while nothing new resolves to it.",
)

roots property 

roots: tuple[Path, ...]

Every directory that is this project: root, then any extra_roots.

name_from_root 

name_from_root() -> Self

A project with no name of its own is called after its root's directory.

local_path has already refused a root with no name, so there is always one to take. A name two projects both want is refused by ProjectManager.reconcile, which is where the names in use are known.

Source code in octomate/schemas/project.py 
132
133
134
135
136
137
138
139
140
141
@model_validator(mode="after")
def name_from_root(self) -> Self:
    """A project with no name of its own is called after its root's directory.

    `local_path` has already refused a root with no name, so there is always one
    to take. A name two projects both want is refused by
    `ProjectManager.reconcile`, which is where the names in use are known.
    """
    self.name = self.name or self.root.name
    return self

contains 

contains(path: Path) -> bool

Whether path is one of this project's roots, or inside one.

Both sides are resolved, because is_relative_to is lexical: a .. and a symlink are the whole attack surface of a check that compares the string it was handed. A path that does not exist resolves fine, which is what a write creating a file needs.

Source code in octomate/schemas/project.py 
148
149
150
151
152
153
154
155
156
157
def contains(self, path: Path) -> bool:
    """Whether `path` is one of this project's roots, or inside one.

    Both sides are resolved, because `is_relative_to` is lexical: a `..` and a
    symlink are the whole attack surface of a check that compares the string it
    was handed. A path that does not exist resolves fine, which is what a write
    creating a file needs.
    """
    resolved = path.resolve()
    return any(resolved.is_relative_to(root.resolve()) for root in self.roots)

local_path 

local_path(value: str | Path) -> Path

A project root, normalized: an absolute local path with a name to go by.

~ is expanded here for the reason ConfigPath exists: pydantic keeps ~/... literal, and Path("~/x").resolve() yields <cwd>/~/x — a root like that matches nothing, silently. A relative root is made absolute for the same reason, since it would otherwise hang off whatever directory Octomate was started in.

A url is refused rather than mangled: Path("file:///srv/x") is a perfectly real relative path, and it matches nothing. That check runs before the Path is built because PurePath collapses :// to :/, and the scheme stops being visible.

Having a name is the one thing a root needs that a directory does not, so that everything downstream gets a directory it can call the project by.

The directory is deliberately not required to exist. A row outlives the directory it was declared at — reconciliation disables it rather than deleting it — and this type is what rehydrates every row: refusing a dead path here would make one deleted checkout enough to stop the registry loading at all.

Source code in octomate/schemas/project.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
def local_path(value: str | Path) -> Path:
    """A project root, normalized: an absolute local path with a name to go by.

    `~` is expanded here for the reason `ConfigPath` exists: pydantic keeps `~/...`
    literal, and `Path("~/x").resolve()` yields ``<cwd>/~/x`` — a root like that
    matches nothing, silently. A relative root is made absolute for the same reason,
    since it would otherwise hang off whatever directory Octomate was started in.

    A url is refused rather than mangled: `Path("file:///srv/x")` is a perfectly real
    relative path, and it matches nothing. That check runs before the `Path` is built
    because `PurePath` collapses `://` to `:/`, and the scheme stops being visible.

    Having a name is the one thing a root needs that a directory does not, so that
    everything downstream gets a directory it can call the project by.

    The directory is deliberately not required to exist. A row outlives the directory
    it was declared at — reconciliation disables it rather than deleting it — and this
    type is what rehydrates every row: refusing a dead path here would make one
    deleted checkout enough to stop the registry loading at all.
    """
    if "://" in str(value):
        raise ValueError(f"{value!r} is a url; a root is a plain local path")
    path = Path(value).expanduser().absolute()
    if not path.name:
        raise ValueError(f"{path} is the filesystem root, not a project")
    return path

octomate.schemas.mcp

MCP installations: their auth variants, install requests, authorization results and summaries.

McpAuthorizationResult 

McpAuthorizationResult = Annotated[
    McpAuthorizationStatus
    | McpBrowserAuthorizationPending
    | McpDeviceAuthorizationPending,
    Field(discriminator="status"),
]

McpVariant 

McpVariant = NoAuthMcp | BearerMcp | OAuthMcp

McpAppToolMeta pydantic-model

Bases: BaseModel

Extract the app URI, including the SDK's legacy spelling.

Unrelated metadata is ignored here and preserved on the original Tool.

Fields:

  • resource_uri (Annotated[AnyUrl, UrlConstraints(allowed_schemes=['ui'])] | None)
Source code in octomate/schemas/mcp.py 
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
class McpAppToolMeta(BaseModel):
    """Extract the app URI, including the SDK's legacy spelling.

    Unrelated metadata is ignored here and preserved on the original Tool.
    """

    resource_uri: Annotated[AnyUrl, UrlConstraints(allowed_schemes=["ui"])] | None = (
        Field(
            default=None,
            validation_alias=AliasChoices(
                AliasPath("ui", "resourceUri"), "ui/resourceUri"
            ),
            description="The UI resource advertised by a tool, when present.",
        )
    )

resource_uri pydantic-field 

resource_uri: (
    Annotated[
        AnyUrl, UrlConstraints(allowed_schemes=["ui"])
    ]
    | None
) = None

The UI resource advertised by a tool, when present.

McpAppInvocation pydantic-model

Bases: BaseModel

A completed MCP app call, retained independently of its installation.

The tool declaration and complete result are snapshots: replay needs neither another tool call nor a namespace lookup that could resolve a replacement installation. The UI resource itself is fetched under the viewer's access.

Fields:

Source code in octomate/schemas/mcp.py 
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
class McpAppInvocation(BaseModel):
    """A completed MCP app call, retained independently of its installation.

    The tool declaration and complete result are snapshots: replay needs neither
    another tool call nor a namespace lookup that could resolve a replacement
    installation. The UI resource itself is fetched under the viewer's access.
    """

    installation_id: UUID7 = Field(
        description="The original installation's identity, retained after uninstall."
    )
    namespace: str = Field(description="The installation's namespace at execution.")
    provider_name: str = Field(
        description="The installation's display name at execution."
    )
    tool: Tool = Field(description="The upstream tool declaration at execution.")
    resource_uri: Annotated[AnyUrl, UrlConstraints(allowed_schemes=["ui"])] = Field(
        description="The UI resource declared by the tool, read through its installation.",
    )
    arguments: dict[str, JsonValue] = Field(
        description="Original arguments passed to the upstream tool, without clipping."
    )
    result: CallToolResult = Field(
        description="Complete upstream result, including structured content and metadata."
    )

installation_id pydantic-field 

installation_id: UUID7

The original installation's identity, retained after uninstall.

namespace pydantic-field 

namespace: str

The installation's namespace at execution.

provider_name pydantic-field 

provider_name: str

The installation's display name at execution.

tool pydantic-field 

tool: Tool

The upstream tool declaration at execution.

resource_uri pydantic-field 

resource_uri: AnyUrl

The UI resource declared by the tool, read through its installation.

arguments pydantic-field 

arguments: dict[str, JsonValue]

Original arguments passed to the upstream tool, without clipping.

result pydantic-field 

result: CallToolResult

Complete upstream result, including structured content and metadata.

McpAppResourceCSP

Bases: ResourceCSP

Declared origins, restricted to HTTP(S) and network-only WS(S) sources.

Subdomain wildcards are supported. CSP keywords, credentials, paths, queries, fragments, and directive separators are refused before header construction.

Source code in octomate/schemas/mcp.py 
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
class McpAppResourceCSP(ResourceCSP):
    """Declared origins, restricted to HTTP(S) and network-only WS(S) sources.

    Subdomain wildcards are supported. CSP keywords, credentials, paths, queries,
    fragments, and directive separators are refused before header construction.
    """

    @field_validator(
        "connect_domains", "resource_domains", "frame_domains", "base_uri_domains"
    )
    @classmethod
    def validate_domains(
        cls, domains: list[str] | None, info: ValidationInfo
    ) -> list[str] | None:
        if domains is None:
            return None
        schemes = {"http", "https"}
        if info.field_name == "connect_domains":
            schemes.update({"ws", "wss"})
        validated = []
        for domain in domains:
            if any(char.isspace() or char in ";,'\"\\" for char in domain):
                raise ValueError("CSP sources must be HTTP(S) or WS(S) origins")
            url = AnyUrl(domain)
            if (
                url.scheme not in schemes
                or url.host is None
                or "*" in url.host.removeprefix("*.")
                or url.username is not None
                or url.password is not None
                or url.path not in {None, "/"}
                or url.query is not None
                or url.fragment is not None
            ):
                raise ValueError("CSP sources must be HTTP(S) or WS(S) origins")
            validated.append(str(url).removesuffix("/"))
        return validated

validate_domains classmethod 

validate_domains(
    domains: list[str] | None, info: ValidationInfo
) -> list[str] | None
Source code in octomate/schemas/mcp.py 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
@field_validator(
    "connect_domains", "resource_domains", "frame_domains", "base_uri_domains"
)
@classmethod
def validate_domains(
    cls, domains: list[str] | None, info: ValidationInfo
) -> list[str] | None:
    if domains is None:
        return None
    schemes = {"http", "https"}
    if info.field_name == "connect_domains":
        schemes.update({"ws", "wss"})
    validated = []
    for domain in domains:
        if any(char.isspace() or char in ";,'\"\\" for char in domain):
            raise ValueError("CSP sources must be HTTP(S) or WS(S) origins")
        url = AnyUrl(domain)
        if (
            url.scheme not in schemes
            or url.host is None
            or "*" in url.host.removeprefix("*.")
            or url.username is not None
            or url.password is not None
            or url.path not in {None, "/"}
            or url.query is not None
            or url.fragment is not None
        ):
            raise ValueError("CSP sources must be HTTP(S) or WS(S) origins")
        validated.append(str(url).removesuffix("/"))
    return validated

McpAppResourceUI

Bases: AppConfig

Resource UI metadata; requested permissions and domains are not grants.

Source code in octomate/schemas/mcp.py 
120
121
122
123
124
125
class McpAppResourceUI(AppConfig):
    """Resource UI metadata; requested permissions and domains are not grants."""

    csp: McpAppResourceCSP | None = Field(
        default=None, description="Validated declared origins for sandbox policy."
    )

csp class-attribute instance-attribute 

csp: McpAppResourceCSP | None = Field(
    default=None,
    description="Validated declared origins for sandbox policy.",
)

McpAppResource pydantic-model

Bases: BaseModel

A single matching MCP Apps HTML resource, decoded for the sandbox.

HTML is untrusted and must only be rendered by the isolated sandbox. Resource metadata describes requests; the host still applies its own sandbox policy.

Fields:

Validators:

Source code in octomate/schemas/mcp.py 
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
class McpAppResource(BaseModel):
    """A single matching MCP Apps HTML resource, decoded for the sandbox.

    HTML is untrusted and must only be rendered by the isolated sandbox. Resource
    metadata describes requests; the host still applies its own sandbox policy.
    """

    uri: Annotated[AnyUrl, UrlConstraints(allowed_schemes=["ui"])] = Field(
        description="The saved invocation's UI resource URI."
    )
    html: str = Field(
        min_length=1, description="Resource text or strictly decoded UTF-8 HTML."
    )
    ui: McpAppResourceUI = Field(
        default_factory=McpAppResourceUI, description="The resource's UI metadata."
    )

    @field_validator("html")
    @classmethod
    def validate_html(cls, value: str) -> str:
        value.encode("utf-8")
        if not value.strip():
            raise ValueError("MCP app HTML must not be empty")
        return value

uri pydantic-field 

uri: AnyUrl

The saved invocation's UI resource URI.

html pydantic-field 

html: str

Resource text or strictly decoded UTF-8 HTML.

ui pydantic-field 

ui: McpAppResourceUI

The resource's UI metadata.

validate_html pydantic-validator 

validate_html(value: str) -> str
Source code in octomate/schemas/mcp.py 
145
146
147
148
149
150
151
@field_validator("html")
@classmethod
def validate_html(cls, value: str) -> str:
    value.encode("utf-8")
    if not value.strip():
        raise ValueError("MCP app HTML must not be empty")
    return value

NoAuth pydantic-model

Bases: BaseModel

No authentication.

Fields:

  • kind (Literal['none'])
Source code in octomate/schemas/mcp.py 
154
155
156
157
class NoAuth(BaseModel):
    """No authentication."""

    kind: Literal["none"] = "none"

kind pydantic-field 

kind: Literal['none'] = 'none'

BearerAuth pydantic-model

Bases: BaseModel

A static bearer token.

Fields:

Source code in octomate/schemas/mcp.py 
160
161
162
163
164
class BearerAuth(BaseModel):
    """A static bearer token."""

    kind: Literal["bearer"] = "bearer"
    token: SecretStr = Field(min_length=1, max_length=16384, repr=False)

kind pydantic-field 

kind: Literal['bearer'] = 'bearer'

token pydantic-field 

token: SecretStr

OAuth pydantic-model

Bases: BaseModel

OAuth, authorized by the installing user.

Config:

  • extra: forbid

Fields:

  • kind (Literal['oauth'])
Source code in octomate/schemas/mcp.py 
167
168
169
170
171
class OAuth(BaseModel):
    """OAuth, authorized by the installing user."""

    model_config = ConfigDict(extra="forbid")
    kind: Literal["oauth"] = "oauth"

model_config class-attribute instance-attribute 

model_config = ConfigDict(extra='forbid')

kind pydantic-field 

kind: Literal['oauth'] = 'oauth'

McpAuthorizationStatus pydantic-model

Bases: BaseModel

The stored grant's status, or null when there is none.

Fields:

  • status (OAuthConnectionStatus | Literal[None])
Source code in octomate/schemas/mcp.py 
174
175
176
177
class McpAuthorizationStatus(BaseModel):
    """The stored grant's status, or null when there is none."""

    status: OAuthConnectionStatus | Literal[None]

status pydantic-field 

status: OAuthConnectionStatus | Literal[None]

McpBrowserAuthorizationPending pydantic-model

Bases: BaseModel

A browser authorization is waiting on the user.

Fields:

  • status (Literal['pending_browser'])
Source code in octomate/schemas/mcp.py 
180
181
182
183
class McpBrowserAuthorizationPending(BaseModel):
    """A browser authorization is waiting on the user."""

    status: Literal["pending_browser"] = "pending_browser"

status pydantic-field 

status: Literal['pending_browser'] = 'pending_browser'

McpDeviceAuthorizationPending pydantic-model

Bases: BaseModel

A device authorization is waiting on the user; poll again after the interval.

Fields:

Source code in octomate/schemas/mcp.py 
186
187
188
189
190
191
class McpDeviceAuthorizationPending(BaseModel):
    """A device authorization is waiting on the user; poll again after the
    interval."""

    status: Literal["pending_device"] = "pending_device"
    retry_after_seconds: int = Field(ge=1)

status pydantic-field 

status: Literal['pending_device'] = 'pending_device'

retry_after_seconds pydantic-field 

retry_after_seconds: int

McpInstallRequest pydantic-model

Bases: BaseModel

What installing an MCP takes: a name, a namespace, an HTTPS endpoint, and either an auth choice or the tentacle that supplies it.

Fields:

Validators:

Source code in octomate/schemas/mcp.py 
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
class McpInstallRequest(BaseModel):
    """What installing an MCP takes: a name, a namespace, an HTTPS endpoint, and
    either an auth choice or the tentacle that supplies it."""

    name: str = Field(min_length=1, max_length=100)
    namespace: str = Field(pattern=r"^[a-z][a-z0-9_-]{0,63}$")
    url: Annotated[HttpsUrl, UrlConstraints(max_length=2083)]
    auth: Annotated[NoAuth | BearerAuth | OAuth, Field(discriminator="kind")] | None = (
        None
    )
    tentacle_id: str | None = Field(
        default=None,
        min_length=1,
        description="Configured tentacle supplying the MCP and its authentication.",
    )

    @model_validator(mode="after")
    def public_endpoint(self) -> Self:
        if self.tentacle_id is not None and self.auth is not None:
            raise ValueError("A tentacle supplies its own authentication")
        if (
            self.url.username
            or self.url.password
            or self.url.query
            or self.url.fragment
        ):
            raise ValueError(
                "MCP endpoints require HTTPS without URL credentials, query, or fragment"
            )
        return self

name pydantic-field 

name: str

namespace pydantic-field 

namespace: str

url pydantic-field 

url: HttpsUrl

auth pydantic-field 

auth: (
    Annotated[
        NoAuth | BearerAuth | OAuth,
        Field(discriminator="kind"),
    ]
    | None
) = None

tentacle_id pydantic-field 

tentacle_id: str | None = None

Configured tentacle supplying the MCP and its authentication.

public_endpoint pydantic-validator 

public_endpoint() -> Self
Source code in octomate/schemas/mcp.py 
218
219
220
221
222
223
224
225
226
227
228
229
230
231
@model_validator(mode="after")
def public_endpoint(self) -> Self:
    if self.tentacle_id is not None and self.auth is not None:
        raise ValueError("A tentacle supplies its own authentication")
    if (
        self.url.username
        or self.url.password
        or self.url.query
        or self.url.fragment
    ):
        raise ValueError(
            "MCP endpoints require HTTPS without URL credentials, query, or fragment"
        )
    return self

Mcp

Bases: BaseTransmuter

A user's MCP installation, polymorphic on how it authenticates.

Source code in octomate/schemas/mcp.py 
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
@sqlalchemy_materia.bless(mcp_models.Mcp)
class Mcp(BaseTransmuter):
    """A user's MCP installation, polymorphic on how it authenticates."""

    model_config = ConfigDict(from_attributes=True)

    id: Annotated[UUID7, Identity] = Field(default_factory=uuid7, frozen=True)
    user_id: UUID7 = Field(frozen=True, exclude=True)
    name: str
    namespace: str = Field(frozen=True)
    url: str = Field(frozen=True)
    instructions: str = ""
    tentacle_id: str | None = None
    enabled: bool = True
    created_at: AwareDatetime = Field(default_factory=lambda: datetime.now(UTC))
    updated_at: AwareDatetime = Field(default_factory=lambda: datetime.now(UTC))

    @property
    def summary(self) -> McpServerSummary:
        match self:
            case NoAuthMcp() | BearerMcp() | OAuthMcp():
                auth_kind = self.auth_kind
            case _:
                raise TypeError("MCP info requires a concrete authentication type")
        return McpServerSummary(
            id=self.id,
            namespace=self.namespace,
            name=self.name,
            enabled=self.enabled,
            auth_kind=auth_kind,
        )

model_config class-attribute instance-attribute 

model_config = ConfigDict(from_attributes=True)

id class-attribute instance-attribute 

id: Annotated[UUID7, Identity] = Field(
    default_factory=uuid7, frozen=True
)

user_id class-attribute instance-attribute 

user_id: UUID7 = Field(frozen=True, exclude=True)

name instance-attribute 

name: str

namespace class-attribute instance-attribute 

namespace: str = Field(frozen=True)

url class-attribute instance-attribute 

url: str = Field(frozen=True)

instructions class-attribute instance-attribute 

instructions: str = ''

tentacle_id class-attribute instance-attribute 

tentacle_id: str | None = None

enabled class-attribute instance-attribute 

enabled: bool = True

created_at class-attribute instance-attribute 

created_at: AwareDatetime = Field(
    default_factory=lambda: datetime.now(UTC)
)

updated_at class-attribute instance-attribute 

updated_at: AwareDatetime = Field(
    default_factory=lambda: datetime.now(UTC)
)

summary property 

summary: McpServerSummary

NoAuthMcp

Bases: Mcp

An installation that sends no credentials.

Source code in octomate/schemas/mcp.py 
267
268
269
270
271
@sqlalchemy_materia.bless(mcp_models.NoAuthMcp)
class NoAuthMcp(Mcp):
    """An installation that sends no credentials."""

    auth_kind: Literal["none"] = "none"

auth_kind class-attribute instance-attribute 

auth_kind: Literal['none'] = 'none'

BearerMcp

Bases: Mcp

An installation authenticating with an encrypted bearer token.

Source code in octomate/schemas/mcp.py 
274
275
276
277
278
279
@sqlalchemy_materia.bless(mcp_models.BearerMcp)
class BearerMcp(Mcp):
    """An installation authenticating with an encrypted bearer token."""

    auth_kind: Literal["bearer"] = "bearer"
    encrypted_token: bytes = Field(min_length=1, exclude=True, repr=False)

auth_kind class-attribute instance-attribute 

auth_kind: Literal['bearer'] = 'bearer'

encrypted_token class-attribute instance-attribute 

encrypted_token: bytes = Field(
    min_length=1, exclude=True, repr=False
)

OAuthMcp

Bases: Mcp

An installation authenticating with the user's OAuth grant.

Source code in octomate/schemas/mcp.py 
282
283
284
285
286
@sqlalchemy_materia.bless(mcp_models.OAuthMcp)
class OAuthMcp(Mcp):
    """An installation authenticating with the user's OAuth grant."""

    auth_kind: Literal["oauth"] = "oauth"

auth_kind class-attribute instance-attribute 

auth_kind: Literal['oauth'] = 'oauth'

McpTentacleInfo pydantic-model

Bases: BaseModel

A configured MCP tentacle as offered for installation.

Fields:

Source code in octomate/schemas/mcp.py 
289
290
291
292
293
294
295
class McpTentacleInfo(BaseModel):
    """A configured MCP tentacle as offered for installation."""

    id: str
    name: str
    url: str
    auth_kind: Literal["none", "bearer", "oauth"]

id pydantic-field 

id: str

name pydantic-field 

name: str

url pydantic-field 

url: str

auth_kind pydantic-field 

auth_kind: Literal['none', 'bearer', 'oauth']

McpOAuthSummary pydantic-model

Bases: BaseModel

An OAuth installation's grant status and the flows it supports.

Fields:

  • status (OAuthConnectionStatus | Literal['pending_browser', 'pending_device', None])
  • flows (list[OAuthFlowKind])
Source code in octomate/schemas/mcp.py 
298
299
300
301
302
303
304
305
306
class McpOAuthSummary(BaseModel):
    """An OAuth installation's grant status and the flows it supports."""

    status: (
        OAuthConnectionStatus | Literal["pending_browser", "pending_device", None]
    ) = Field(
        description="Stored grant or live authorization status; null means neither exists."
    )
    flows: list[OAuthFlowKind]

status pydantic-field 

status: (
    OAuthConnectionStatus
    | Literal["pending_browser", "pending_device", None]
)

Stored grant or live authorization status; null means neither exists.

flows pydantic-field 

flows: list[OAuthFlowKind]

McpServerSummary pydantic-model

Bases: BaseModel

An installation as listed: identity, state and auth kind.

Fields:

Source code in octomate/schemas/mcp.py 
309
310
311
312
313
314
315
316
317
class McpServerSummary(BaseModel):
    """An installation as listed: identity, state and auth kind."""

    id: UUID7
    namespace: str
    name: str
    enabled: bool
    auth_kind: Literal["none", "bearer", "oauth"]
    oauth: McpOAuthSummary | None = None

id pydantic-field 

id: UUID7

namespace pydantic-field 

namespace: str

name pydantic-field 

name: str

enabled pydantic-field 

enabled: bool

auth_kind pydantic-field 

auth_kind: Literal['none', 'bearer', 'oauth']

oauth pydantic-field 

oauth: McpOAuthSummary | None = None

McpToolCatalog pydantic-model

Bases: BaseModel

The tools and instructions an installation currently serves.

Fields:

Source code in octomate/schemas/mcp.py 
320
321
322
323
324
325
326
327
328
329
class McpToolCatalog(BaseModel):
    """The tools and instructions an installation currently serves."""

    instructions: str = Field(description="The selected provider's tool instructions.")
    tools: list[Tool] = Field(description="The selected provider's MCP tool schemas.")

    @cached_property
    def tools_by_name(self) -> dict[str, Tool]:
        """Index this discovery snapshot once; refresh creates a new catalog."""
        return {tool.name: tool for tool in self.tools}

instructions pydantic-field 

instructions: str

The selected provider's tool instructions.

tools pydantic-field 

tools: list[Tool]

The selected provider's MCP tool schemas.

tools_by_name cached property 

tools_by_name: dict[str, Tool]

Index this discovery snapshot once; refresh creates a new catalog.

octomate.schemas.oauth

OAuth machinery: the cipher, flow payloads, callback transports, and the persisted operation and connection.

OAUTH_START_PATH module-attribute 

OAUTH_START_PATH = (
    "/oauth/{connector_id}/start/{operation_id}"
)

OAUTH_CALLBACK_PATH module-attribute 

OAUTH_CALLBACK_PATH = '/oauth/{connector_id}/callback'

OAuthStartResult module-attribute 

OAuthStartResult = DeviceAuthorization | AuthorizationLink

OAuthCipher

Encrypts short-lived operation secrets and durable OAuth tokens at rest.

Source code in octomate/schemas/oauth.py 
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
class OAuthCipher:
    """Encrypts short-lived operation secrets and durable OAuth tokens at rest."""

    def __init__(self, encryption_key: SecretStr) -> None:
        try:
            key = b64decode(
                encryption_key.get_secret_value(),
                altchars=b"-_",
                validate=True,
            )
        except ValueError as error:
            raise ValueError("oauth.encryption_key must be URL-safe base64") from error
        if len(key) != 32:
            raise ValueError("oauth.encryption_key must encode exactly 32 bytes")
        self.aes = AESGCM(key)

    def encrypt(self, value: str, *, context: str) -> bytes:
        nonce = urandom(12)
        return nonce + self.aes.encrypt(nonce, value.encode(), context.encode())

    def decrypt(self, value: bytes, *, context: str) -> str:
        nonce, ciphertext = value[:12], value[12:]
        return self.aes.decrypt(nonce, ciphertext, context.encode()).decode()

aes instance-attribute 

aes = AESGCM(key)

encrypt 

encrypt(value: str, *, context: str) -> bytes
Source code in octomate/schemas/oauth.py 
52
53
54
def encrypt(self, value: str, *, context: str) -> bytes:
    nonce = urandom(12)
    return nonce + self.aes.encrypt(nonce, value.encode(), context.encode())

decrypt 

decrypt(value: bytes, *, context: str) -> str
Source code in octomate/schemas/oauth.py 
56
57
58
def decrypt(self, value: bytes, *, context: str) -> str:
    nonce, ciphertext = value[:12], value[12:]
    return self.aes.decrypt(nonce, ciphertext, context.encode()).decode()

OAuthFlowContext dataclass

The manager's owner-bound input to a connector strategy.

OAuthManager creates it only after UserManager resolves the initiating channel profile to a registered user. The operation id correlates later device polling or browser callbacks; it never selects or overrides the owning user.

Source code in octomate/schemas/oauth.py 
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
@dataclass(frozen=True)
class OAuthFlowContext:
    """The manager's owner-bound input to a connector strategy.

    ``OAuthManager`` creates it only after ``UserManager`` resolves the initiating
    channel profile to a registered user. The operation id correlates later device
    polling or browser callbacks; it never selects or overrides the owning user.
    """

    operation_id: UUID7
    connector_id: str
    user: User
    profile: UserProfile | None
    mcp_id: UUID7 | None = None
    interval_seconds: int | None = None  # Current device polling interval.

operation_id instance-attribute 

operation_id: UUID7

connector_id instance-attribute 

connector_id: str

user instance-attribute 

user: User

profile instance-attribute 

profile: UserProfile | None

mcp_id class-attribute instance-attribute 

mcp_id: UUID7 | None = None

interval_seconds class-attribute instance-attribute 

interval_seconds: int | None = None

OAuthDiscoveryState pydantic-model

Bases: BaseModel

Discovered OAuth server metadata and client registration retained for reuse.

Fields:

Source code in octomate/schemas/oauth.py 
78
79
80
81
82
83
84
85
86
87
class OAuthDiscoveryState(BaseModel):
    """Discovered OAuth server metadata and client registration retained for reuse."""

    resource: HttpsUrl | None = Field(
        default=None,
        description="OAuth resource indicator, when the server requires one.",
    )
    metadata: OAuthMetadata
    client: OAuthClientInformationFull = Field(repr=False)
    scope: str | None = None

resource pydantic-field 

resource: HttpsUrl | None = None

OAuth resource indicator, when the server requires one.

metadata pydantic-field 

metadata: OAuthMetadata

client pydantic-field 

client: OAuthClientInformationFull

scope pydantic-field 

scope: str | None = None

DeviceAuthorizationResponse pydantic-model

Bases: BaseModel

The authorization server's response when a device flow starts.

A DeviceAuthorizationFlow returns this internal value to the manager. The manager adds its operation id before the verification instructions leave the OAuth boundary; the future completion step polls with server-side operation state.

Fields:

Source code in octomate/schemas/oauth.py 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
class DeviceAuthorizationResponse(BaseModel):
    """The authorization server's response when a device flow starts.

    A ``DeviceAuthorizationFlow`` returns this internal value to the manager. The manager
    adds its operation id before the verification instructions leave the OAuth
    boundary; the future completion step polls with server-side operation state.
    """

    verification_uri: HttpsUrl
    verification_uri_complete: HttpsUrl | None = None
    device_code: SecretStr = Field(exclude=True, repr=False)
    user_code: SecretStr = Field(repr=False)
    expires_at: AwareDatetime
    interval_seconds: int = Field(ge=1)

verification_uri pydantic-field 

verification_uri: HttpsUrl

verification_uri_complete pydantic-field 

verification_uri_complete: HttpsUrl | None = None

device_code pydantic-field 

device_code: SecretStr

user_code pydantic-field 

user_code: SecretStr

expires_at pydantic-field 

expires_at: AwareDatetime

interval_seconds pydantic-field 

interval_seconds: int

AuthorizationRequest pydantic-model

Bases: BaseModel

The provider-facing request that starts authorization-code OAuth.

An AuthorizationCodeFlow builds this after receiving the callback URI. Its URI may contain OAuth state and must not be sent to a channel directly. The selected callback transport stages it and returns an AuthorizationLink.

Fields:

Source code in octomate/schemas/oauth.py 
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
class AuthorizationRequest(BaseModel):
    """The provider-facing request that starts authorization-code OAuth.

    An ``AuthorizationCodeFlow`` builds this after receiving the callback URI.
    Its URI may contain OAuth state and must not be sent to a channel directly. The
    selected callback transport stages it and returns an ``AuthorizationLink``.
    """

    authorization_uri: AnyHttpUrl = Field(repr=False)
    code_verifier: SecretStr | None = Field(
        default=None,
        repr=False,
        description="The PKCE verifier whose challenge the authorization URI carries, "
        "returned so the manager can seal it into the operation and spend it at the "
        "token exchange. None for a provider that does not offer PKCE.",
    )
    expires_at: AwareDatetime
    discovery_state: OAuthDiscoveryState | None = Field(default=None, repr=False)

authorization_uri pydantic-field 

authorization_uri: AnyHttpUrl

code_verifier pydantic-field 

code_verifier: SecretStr | None = None

The PKCE verifier whose challenge the authorization URI carries, returned so the manager can seal it into the operation and spend it at the token exchange. None for a provider that does not offer PKCE.

expires_at pydantic-field 

expires_at: AwareDatetime

discovery_state pydantic-field 

discovery_state: OAuthDiscoveryState | None = None

DeviceAuthorization pydantic-model

Bases: BaseModel

Device verification instructions returned by the manager.

The channel presents the verification URI and user code to the initiating user. Its operation id lets the later polling and confirmation steps resume the exact owner-bound authorization without accepting a user id from the caller.

Fields:

Source code in octomate/schemas/oauth.py 
126
127
128
129
130
131
132
133
134
135
136
137
138
139
class DeviceAuthorization(BaseModel):
    """Device verification instructions returned by the manager.

    The channel presents the verification URI and user code to the initiating user.
    Its operation id lets the later polling and confirmation steps resume the exact
    owner-bound authorization without accepting a user id from the caller.
    """

    operation_id: UUID7
    verification_uri: HttpsUrl
    verification_uri_complete: HttpsUrl | None = None
    user_code: SecretStr = Field(repr=False)
    expires_at: AwareDatetime
    interval_seconds: int = Field(ge=1)

operation_id pydantic-field 

operation_id: UUID7

verification_uri pydantic-field 

verification_uri: HttpsUrl

verification_uri_complete pydantic-field 

verification_uri_complete: HttpsUrl | None = None

user_code pydantic-field 

user_code: SecretStr

expires_at pydantic-field 

expires_at: AwareDatetime

interval_seconds pydantic-field 

interval_seconds: int

OAuthPending pydantic-model

Bases: BaseModel

A device token exchange that the authorization server has not completed.

The manager returns this when the user has not yet approved the device code. The channel can ask the same owner to confirm again after retry_after_seconds.

Fields:

Source code in octomate/schemas/oauth.py 
142
143
144
145
146
147
148
149
class OAuthPending(BaseModel):
    """A device token exchange that the authorization server has not completed.

    The manager returns this when the user has not yet approved the device code.
    The channel can ask the same owner to confirm again after ``retry_after_seconds``.
    """

    retry_after_seconds: int = Field(ge=1)

retry_after_seconds pydantic-field 

retry_after_seconds: int

OAuthGrant pydantic-model

Bases: BaseModel

Provider credentials returned after the user authorizes an operation.

A device flow or authorization-code flow returns this only to OAuthManager. The manager encrypts its secrets before persisting the owner-bound connection.

Fields:

Source code in octomate/schemas/oauth.py 
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
class OAuthGrant(BaseModel):
    """Provider credentials returned after the user authorizes an operation.

    A device flow or authorization-code flow returns this only to ``OAuthManager``.
    The manager encrypts its secrets before persisting the owner-bound connection.
    """

    access_token: SecretStr = Field(repr=False)
    refresh_token: SecretStr | None = Field(default=None, repr=False)
    token_type: str = "bearer"
    scopes: list[str] = Field(default_factory=list)
    subject: str | None = None
    account_label: str | None = None
    expires_at: AwareDatetime | None = None
    discovery_state: OAuthDiscoveryState | None = Field(
        default=None, repr=False, exclude=True
    )

access_token pydantic-field 

access_token: SecretStr

refresh_token pydantic-field 

refresh_token: SecretStr | None = None

token_type pydantic-field 

token_type: str = 'bearer'

scopes pydantic-field 

scopes: list[str]

subject pydantic-field 

subject: str | None = None

account_label pydantic-field 

account_label: str | None = None

expires_at pydantic-field 

expires_at: AwareDatetime | None = None

discovery_state pydantic-field 

discovery_state: OAuthDiscoveryState | None = None

Bases: BaseModel

The safe authorization-code link returned to the channel.

The link contains only the operation UUID. Opening it makes the direct or relay transport retrieve the staged provider request, redirect the browser, and return the eventual callback to the manager's shared completion boundary.

Fields:

Source code in octomate/schemas/oauth.py 
171
172
173
174
175
176
177
178
179
180
181
class AuthorizationLink(BaseModel):
    """The safe authorization-code link returned to the channel.

    The link contains only the operation UUID. Opening it makes the direct or relay
    transport retrieve the staged provider request, redirect the browser, and return
    the eventual callback to the manager's shared completion boundary.
    """

    operation_id: UUID7
    authorization_uri: AnyHttpUrl
    expires_at: AwareDatetime

operation_id pydantic-field 

operation_id: UUID7

authorization_uri pydantic-field 

authorization_uri: AnyHttpUrl

expires_at pydantic-field 

expires_at: AwareDatetime

OAuthCallbackTransport

Bases: ABC

Moves authorization-code traffic across deployment ingress.

After the manager establishes the owner, the transport supplies the provider's callback URI and stages the secret-bearing authorization request behind a UUID-only link. Direct HTTP and relay delivery converge on the same later manager completion step.

Source code in octomate/schemas/oauth.py 
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
class OAuthCallbackTransport(ABC):
    """Moves authorization-code traffic across deployment ingress.

    After the manager establishes the owner, the transport supplies the provider's
    callback URI and stages the secret-bearing authorization request behind a
    UUID-only link. Direct HTTP and relay delivery converge on the same later manager
    completion step.
    """

    @property
    @abstractmethod
    def kind(self) -> Literal["direct_http", "relay"]:
        """The deployment transport selected by the connector."""

    @abstractmethod
    async def callback_uri(self, connector_id: str) -> AnyHttpUrl:
        """Return the provider-facing callback URI for this connector."""

    @abstractmethod
    async def prepare_authorization(
        self,
        context: OAuthFlowContext,
        authorization_uri: AnyHttpUrl,
    ) -> AnyHttpUrl:
        """Stage a provider URI and return the UUID-only link sent to the user."""

kind abstractmethod property 

kind: Literal['direct_http', 'relay']

The deployment transport selected by the connector.

callback_uri abstractmethod async 

callback_uri(connector_id: str) -> AnyHttpUrl

Return the provider-facing callback URI for this connector.

Source code in octomate/schemas/oauth.py 
198
199
200
@abstractmethod
async def callback_uri(self, connector_id: str) -> AnyHttpUrl:
    """Return the provider-facing callback URI for this connector."""

prepare_authorization abstractmethod async 

prepare_authorization(
    context: OAuthFlowContext, authorization_uri: AnyHttpUrl
) -> AnyHttpUrl

Stage a provider URI and return the UUID-only link sent to the user.

Source code in octomate/schemas/oauth.py 
202
203
204
205
206
207
208
@abstractmethod
async def prepare_authorization(
    self,
    context: OAuthFlowContext,
    authorization_uri: AnyHttpUrl,
) -> AnyHttpUrl:
    """Stage a provider URI and return the UUID-only link sent to the user."""

DirectHttpOAuthCallbackTransport

Bases: OAuthCallbackTransport

Authorization-code transport backed by Octomate's own HTTP routes.

base_uri is where a browser reaches this deployment, which is the only thing the two paths below need to become real URIs. A loopback address serves local development, where the browser and Octomate are the same machine; a tunnel or domain replaces it for anyone else, and nothing but this value changes.

Source code in octomate/schemas/oauth.py 
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
class DirectHttpOAuthCallbackTransport(OAuthCallbackTransport):
    """Authorization-code transport backed by Octomate's own HTTP routes.

    ``base_uri`` is where a browser reaches this deployment, which is the only thing
    the two paths below need to become real URIs. A loopback address serves local
    development, where the browser and Octomate are the same machine; a tunnel or
    domain replaces it for anyone else, and nothing but this value changes.
    """

    def __init__(self, base_uri: AnyHttpUrl) -> None:
        self.base_uri = str(base_uri).rstrip("/")

    @property
    def kind(self) -> Literal["direct_http"]:
        return "direct_http"

    async def callback_uri(self, connector_id: str) -> AnyHttpUrl:
        return AnyHttpUrl(
            self.base_uri + OAUTH_CALLBACK_PATH.format(connector_id=connector_id)
        )

    async def prepare_authorization(
        self,
        context: OAuthFlowContext,
        authorization_uri: AnyHttpUrl,
    ) -> AnyHttpUrl:
        """Return the UUID-only link, ignoring the provider URI it stands in for.

        Staging already happened: the manager sealed that URI into this operation's
        encrypted payload before asking for a link, and the start route reads it back
        from there. Nothing crosses a deployment boundary, which is the whole
        difference between this transport and a relay.
        """
        return AnyHttpUrl(
            self.base_uri
            + OAUTH_START_PATH.format(
                connector_id=context.connector_id,
                operation_id=context.operation_id,
            )
        )

base_uri instance-attribute 

base_uri = str(base_uri).rstrip('/')

kind property 

kind: Literal['direct_http']

callback_uri async 

callback_uri(connector_id: str) -> AnyHttpUrl
Source code in octomate/schemas/oauth.py 
236
237
238
239
async def callback_uri(self, connector_id: str) -> AnyHttpUrl:
    return AnyHttpUrl(
        self.base_uri + OAUTH_CALLBACK_PATH.format(connector_id=connector_id)
    )

prepare_authorization async 

prepare_authorization(
    context: OAuthFlowContext, authorization_uri: AnyHttpUrl
) -> AnyHttpUrl

Return the UUID-only link, ignoring the provider URI it stands in for.

Staging already happened: the manager sealed that URI into this operation's encrypted payload before asking for a link, and the start route reads it back from there. Nothing crosses a deployment boundary, which is the whole difference between this transport and a relay.

Source code in octomate/schemas/oauth.py 
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
async def prepare_authorization(
    self,
    context: OAuthFlowContext,
    authorization_uri: AnyHttpUrl,
) -> AnyHttpUrl:
    """Return the UUID-only link, ignoring the provider URI it stands in for.

    Staging already happened: the manager sealed that URI into this operation's
    encrypted payload before asking for a link, and the start route reads it back
    from there. Nothing crosses a deployment boundary, which is the whole
    difference between this transport and a relay.
    """
    return AnyHttpUrl(
        self.base_uri
        + OAUTH_START_PATH.format(
            connector_id=context.connector_id,
            operation_id=context.operation_id,
        )
    )

RelayOAuthCallbackTransport

Bases: OAuthCallbackTransport

Authorization-code transport backed by an external callback relay.

Source code in octomate/schemas/oauth.py 
262
263
264
265
266
267
class RelayOAuthCallbackTransport(OAuthCallbackTransport):
    """Authorization-code transport backed by an external callback relay."""

    @property
    def kind(self) -> Literal["relay"]:
        return "relay"

kind property 

kind: Literal['relay']

OAuthTokenPayload pydantic-model

Bases: BaseModel

The plaintext token envelope immediately before encryption or after decryption.

Fields:

Source code in octomate/schemas/oauth.py 
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
class OAuthTokenPayload(BaseModel):
    """The plaintext token envelope immediately before encryption or after decryption."""

    flow: OAuthFlowKind = Field(
        description="The issuing flow whose client authentication is used for refresh."
    )
    access_token: SecretStr = Field(repr=False)
    refresh_token: SecretStr | None = Field(default=None, repr=False)
    token_type: str = "bearer"
    # Existing encrypted payloads used the MCP-specific field name.
    discovery_state: OAuthDiscoveryState | None = Field(
        default=None,
        repr=False,
        validation_alias=AliasChoices("discovery_state", "mcp_oauth"),
    )

    @field_serializer("access_token", "refresh_token", when_used="json")
    def serialize_secret(self, value: SecretStr | None) -> str | None:
        return value.get_secret_value() if value is not None else None

flow pydantic-field 

flow: OAuthFlowKind

The issuing flow whose client authentication is used for refresh.

access_token pydantic-field 

access_token: SecretStr

refresh_token pydantic-field 

refresh_token: SecretStr | None = None

token_type pydantic-field 

token_type: str = 'bearer'

discovery_state pydantic-field 

discovery_state: OAuthDiscoveryState | None = None

serialize_secret 

serialize_secret(value: SecretStr | None) -> str | None
Source code in octomate/schemas/oauth.py 
289
290
291
@field_serializer("access_token", "refresh_token", when_used="json")
def serialize_secret(self, value: SecretStr | None) -> str | None:
    return value.get_secret_value() if value is not None else None

DeviceOperationPayload pydantic-model

Bases: BaseModel

The plaintext device-authorization envelope immediately before encryption.

A provider mints a fresh code on every start, so an authorization that is still pending can only be shown to its user again out of what was sealed here when it began — nothing but the operation row survives the turn that started it.

Fields:

Source code in octomate/schemas/oauth.py 
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
class DeviceOperationPayload(BaseModel):
    """The plaintext device-authorization envelope immediately before encryption.

    A provider mints a fresh code on every start, so an authorization that is still
    pending can only be shown to its user again out of what was sealed here when it
    began — nothing but the operation row survives the turn that started it.
    """

    device_code: SecretStr = Field(repr=False)
    user_code: SecretStr = Field(repr=False)
    verification_uri: HttpsUrl
    verification_uri_complete: HttpsUrl | None = None

    @field_serializer("device_code", "user_code", when_used="json")
    def serialize_secret(self, value: SecretStr) -> str:
        return value.get_secret_value()

device_code pydantic-field 

device_code: SecretStr

user_code pydantic-field 

user_code: SecretStr

verification_uri pydantic-field 

verification_uri: HttpsUrl

verification_uri_complete pydantic-field 

verification_uri_complete: HttpsUrl | None = None

serialize_secret 

serialize_secret(value: SecretStr) -> str
Source code in octomate/schemas/oauth.py 
307
308
309
@field_serializer("device_code", "user_code", when_used="json")
def serialize_secret(self, value: SecretStr) -> str:
    return value.get_secret_value()

AuthorizationCodeOperationPayload pydantic-model

Bases: BaseModel

The plaintext authorization-code envelope immediately before encryption.

Everything the browser round trip needs and nothing the browser can be trusted to carry: the state a callback must match, the PKCE verifier the token exchange spends, the redirect URI that exchange has to replay, and the provider URI the start route redirects to. All of it is sealed here because the request that started the authorization is long gone by the time any of it is used.

Fields:

Source code in octomate/schemas/oauth.py 
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
class AuthorizationCodeOperationPayload(BaseModel):
    """The plaintext authorization-code envelope immediately before encryption.

    Everything the browser round trip needs and nothing the browser can be trusted
    to carry: the state a callback must match, the PKCE verifier the token exchange
    spends, the redirect URI that exchange has to replay, and the provider URI the
    start route redirects to. All of it is sealed here because the request that
    started the authorization is long gone by the time any of it is used.
    """

    state: SecretStr = Field(repr=False)
    code_verifier: SecretStr | None = Field(default=None, repr=False)
    callback_uri: AnyHttpUrl
    authorization_uri: AnyHttpUrl = Field(repr=False)
    # Existing encrypted payloads used the MCP-specific field name.
    discovery_state: OAuthDiscoveryState | None = Field(
        default=None,
        repr=False,
        validation_alias=AliasChoices("discovery_state", "mcp_oauth"),
    )

    @field_serializer("state", "code_verifier", when_used="json")
    def serialize_secret(self, value: SecretStr | None) -> str | None:
        return value.get_secret_value() if value is not None else None

state pydantic-field 

state: SecretStr

code_verifier pydantic-field 

code_verifier: SecretStr | None = None

callback_uri pydantic-field 

callback_uri: AnyHttpUrl

authorization_uri pydantic-field 

authorization_uri: AnyHttpUrl

discovery_state pydantic-field 

discovery_state: OAuthDiscoveryState | None = None

serialize_secret 

serialize_secret(value: SecretStr | None) -> str | None
Source code in octomate/schemas/oauth.py 
333
334
335
@field_serializer("state", "code_verifier", when_used="json")
def serialize_secret(self, value: SecretStr | None) -> str | None:
    return value.get_secret_value() if value is not None else None

OAuthOperation

Bases: BaseTransmuter

An authorization in flight: the encrypted flow state, its expiry, and the poll interval of a device flow.

Source code in octomate/schemas/oauth.py 
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
@sqlalchemy_materia.bless(oauth_models.OAuthOperation)
class OAuthOperation(BaseTransmuter):
    """An authorization in flight: the encrypted flow state, its expiry, and the
    poll interval of a device flow."""

    model_config = ConfigDict(from_attributes=True)

    id: Annotated[UUID7, Identity] = Field(default_factory=uuid7, frozen=True)
    user_id: UUID7
    profile_id: UUID7 | None = None
    mcp_id: UUID7 | None = None
    connector_id: str
    encrypted_data: bytes = Field(repr=False)
    expires_at: AwareDatetime
    # How long the user's channel waits between polls, which only a device flow
    # does; an authorization-code operation is finished by its callback and has
    # nothing to poll.
    interval_seconds: int | None = None
    created_at: AwareDatetime = Field(default_factory=lambda: datetime.now(UTC))
    consumed_at: AwareDatetime | None = None

model_config class-attribute instance-attribute 

model_config = ConfigDict(from_attributes=True)

id class-attribute instance-attribute 

id: Annotated[UUID7, Identity] = Field(
    default_factory=uuid7, frozen=True
)

user_id instance-attribute 

user_id: UUID7

profile_id class-attribute instance-attribute 

profile_id: UUID7 | None = None

mcp_id class-attribute instance-attribute 

mcp_id: UUID7 | None = None

connector_id instance-attribute 

connector_id: str

encrypted_data class-attribute instance-attribute 

encrypted_data: bytes = Field(repr=False)

expires_at instance-attribute 

expires_at: AwareDatetime

interval_seconds class-attribute instance-attribute 

interval_seconds: int | None = None

created_at class-attribute instance-attribute 

created_at: AwareDatetime = Field(
    default_factory=lambda: datetime.now(UTC)
)

consumed_at class-attribute instance-attribute 

consumed_at: AwareDatetime | None = None

OAuthConnection

Bases: BaseTransmuter

A stored grant: the encrypted tokens, their scopes and expiry, and the account they belong to.

Source code in octomate/schemas/oauth.py 
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
@sqlalchemy_materia.bless(oauth_models.OAuthConnection)
class OAuthConnection(BaseTransmuter):
    """A stored grant: the encrypted tokens, their scopes and expiry, and the
    account they belong to."""

    model_config = ConfigDict(from_attributes=True)

    id: Annotated[UUID7, Identity] = Field(default_factory=uuid7, frozen=True)
    user_id: UUID7
    connector_id: str
    mcp_id: UUID7 | None = None
    status: OAuthConnectionStatus = "active"
    encrypted_tokens: bytes = Field(repr=False)
    subject: str | None = None
    account_label: str | None = None
    scopes: list[str] = Field(default_factory=list)
    expires_at: AwareDatetime | None = None
    created_at: AwareDatetime = Field(default_factory=lambda: datetime.now(UTC))
    updated_at: AwareDatetime = Field(default_factory=lambda: datetime.now(UTC))

model_config class-attribute instance-attribute 

model_config = ConfigDict(from_attributes=True)

id class-attribute instance-attribute 

id: Annotated[UUID7, Identity] = Field(
    default_factory=uuid7, frozen=True
)

user_id instance-attribute 

user_id: UUID7

connector_id instance-attribute 

connector_id: str

mcp_id class-attribute instance-attribute 

mcp_id: UUID7 | None = None

status class-attribute instance-attribute 

status: OAuthConnectionStatus = 'active'

encrypted_tokens class-attribute instance-attribute 

encrypted_tokens: bytes = Field(repr=False)

subject class-attribute instance-attribute 

subject: str | None = None

account_label class-attribute instance-attribute 

account_label: str | None = None

scopes class-attribute instance-attribute 

scopes: list[str] = Field(default_factory=list)

expires_at class-attribute instance-attribute 

expires_at: AwareDatetime | None = None

created_at class-attribute instance-attribute 

created_at: AwareDatetime = Field(
    default_factory=lambda: datetime.now(UTC)
)

updated_at class-attribute instance-attribute 

updated_at: AwareDatetime = Field(
    default_factory=lambda: datetime.now(UTC)
)

octomate.schemas.todos

Conversation-scoped todos and their write payload.

Todo

Bases: BaseTransmuter

One conversation-scoped todo, keyed by its short ref.

Source code in octomate/schemas/todos.py 
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
@sqlalchemy_materia.bless(todos_models.Todo)
class Todo(BaseTransmuter):
    """One conversation-scoped todo, keyed by its short `ref`."""

    model_config = ConfigDict(from_attributes=True)

    id: Annotated[UUID7, Identity] = Field(default_factory=uuid7, frozen=True)
    conversation_id: UUID7
    ref: str
    content: str
    active_form: str = ""
    status: TodoStatus = "pending"
    position: int = 0
    parent_ref: str | None = None
    depends_on: list[str] = Field(default_factory=list)
    created_at: AwareDatetime = Field(default_factory=now_utc)
    updated_at: AwareDatetime = Field(default_factory=now_utc)

model_config class-attribute instance-attribute 

model_config = ConfigDict(from_attributes=True)

id class-attribute instance-attribute 

id: Annotated[UUID7, Identity] = Field(
    default_factory=uuid7, frozen=True
)

conversation_id instance-attribute 

conversation_id: UUID7

ref instance-attribute 

ref: str

content instance-attribute 

content: str

active_form class-attribute instance-attribute 

active_form: str = ''

status class-attribute instance-attribute 

status: TodoStatus = 'pending'

position class-attribute instance-attribute 

position: int = 0

parent_ref class-attribute instance-attribute 

parent_ref: str | None = None

depends_on class-attribute instance-attribute 

depends_on: list[str] = Field(default_factory=list)

created_at class-attribute instance-attribute 

created_at: AwareDatetime = Field(default_factory=now_utc)

updated_at class-attribute instance-attribute 

updated_at: AwareDatetime = Field(default_factory=now_utc)

TodoWrite

Bases: TypedDict

Write-payload for TodoManager.write_todos — one item to (re)create.

Source code in octomate/schemas/todos.py 
37
38
39
40
41
42
43
44
45
class TodoWrite(TypedDict):
    """Write-payload for `TodoManager.write_todos` — one item to (re)create."""

    content: str
    status: NotRequired[TodoStatus]
    active_form: NotRequired[str]
    ref: NotRequired[str | None]
    parent_ref: NotRequired[str | None]
    depends_on: NotRequired[list[str]]

content instance-attribute 

content: str

status instance-attribute 

status: NotRequired[TodoStatus]

active_form instance-attribute 

active_form: NotRequired[str]

ref instance-attribute 

ref: NotRequired[str | None]

parent_ref instance-attribute 

parent_ref: NotRequired[str | None]

depends_on instance-attribute 

depends_on: NotRequired[list[str]]

octomate.schemas.spills

The persisted tool-output spill.

ToolOutputSpill

Bases: BaseTransmuter

One oversized tool return, kept by handle until retention closes over it.

Source code in octomate/schemas/spills.py 
17
18
19
20
21
22
23
24
25
26
@sqlalchemy_materia.bless(spills_models.ToolOutputSpill)
class ToolOutputSpill(BaseTransmuter):
    """One oversized tool return, kept by handle until retention closes over it."""

    model_config = ConfigDict(from_attributes=True)

    id: Annotated[UUID7, Identity] = Field(default_factory=uuid7, frozen=True)
    handle: str
    payload: bytes
    created_at: AwareDatetime = Field(default_factory=lambda: datetime.now(UTC))

model_config class-attribute instance-attribute 

model_config = ConfigDict(from_attributes=True)

id class-attribute instance-attribute 

id: Annotated[UUID7, Identity] = Field(
    default_factory=uuid7, frozen=True
)

handle instance-attribute 

handle: str

payload instance-attribute 

payload: bytes

created_at class-attribute instance-attribute 

created_at: AwareDatetime = Field(
    default_factory=lambda: datetime.now(UTC)
)