angee.mcp.graphql

Expose curated GraphQL operations as MCP tools, scoped to the request actor.

The reuse seam for the MCP tool layer: instead of re-deriving CRUD, projection, and permission gating per model, an MCP tool runs the same GraphQL operation a browser would, under the agent's REBAC actor — so strawberry's own permission_classes and RebacManager scoping do the authorization.

An addon declares a :class:GraphQLTool per operation and calls :func:register; the compiler introspects the named schema bucket (graphql-core) to derive the tool's input schema, response projection, and operation document, then registers a FastMCP tool whose :meth:_CompiledTool.run executes the operation through :func:execute_under_actor.

Boundary conventions (the agent surface differs from the GraphQL wire):

• ids are the public sqid; a GraphQL id arg/field is exposed as sqid.
• field names are snake_case for the agent; the compiler uses the schema's actual wire name, whether camelCase or Hasura snake_case.
• a single input object (createNote(data:) / insert_notes_one(object:)) is flattened to top-level tool args.
• an offset-paginated list exposes limit → pagination.limit and projects results; a Hasura list uses top-level limit and returns rows directly.

execute_under_actor

async def execute_under_actor(
        schema: str,
        document: str,
        variables: dict[str, Any] | None = None) -> dict[str, Any]

Execute document against the named schema bucket under the ambient actor.

Returns the operation's data. Raises the first GraphQL error so the MCP layer surfaces it as a tool error rather than returning a partial result. The actor is whatever rebac.current_actor() holds — set per call by :class:angee.mcp.middleware.ActorMiddleware; the schema's RebacExtension opens its own scopes and pins every queryset to that actor, so no Django request is needed.

Runs execute_sync in a thread: it matches the sync Django GraphQL view the browser uses, and the crud delete resolver does sync ORM that isn't async-safe. sync_to_async (thread-sensitive) copies the ambient actor ContextVar into the sync thread.

GraphQLTool

@dataclass(frozen=True)
class GraphQLTool()

Declaration of one MCP tool backed by a GraphQL operation.

operation is the root field name in the schema bucket; name is the MCP tool name. fields is the response projection in snake_case (sqid selects the node's public id). Each entry is a scalar leaf or a (name, (child, ...)) branch that projects a nested object/list one or two levels deep (see :data:ProjectionSpec). The compiler derives the input schema and document from introspection; the hints below name the input args the tool drives: flatten lifts an input object's fields to top-level args, id_arg exposes a scalar GraphQL id arg as sqid, limit_arg maps a top-level int to pagination.limit for an offset-paginated list, args passes named root arguments straight through as top-level tool inputs (scalars, enums, or lists thereof — for operations whose inputs are bare arguments rather than one input object), and fixed injects constant GraphQL arguments the agent never sees (e.g. confirm on a delete).

register_graphql_tools

def register_graphql_tools(server: Any, specs: list[GraphQLTool]) -> None

Compile each spec against its schema bucket and add it to the FastMCP server.

_Projection

class _Projection(BaseModel)

One node in a tool's response projection tree (object nesting ≤ 2 levels).

Resolved once at compile time so the runtime carries no closures and no live schema reference: wire is the schema field name, key the agent-facing snake_case name, is_id marks the public-id (sqid) translation, is_list a list-valued branch, and children the nested projection for an object/list branch (empty for a scalar leaf). The node owns its own document fragment, output schema, and value extraction.

selection

def selection() -> str

Render this node's GraphQL selection: a leaf wire or wire { ... }.

json_schema

def json_schema(node: Any) -> dict[str, Any]

Build this node's JSON output schema, descending node for child types.

value

def value(row: dict[str, Any]) -> Any

Extract this node's projected value from a parent GraphQL row.

project_row

def project_row(projections: tuple[_Projection, ...],
                row: dict[str, Any]) -> dict[str, Any]

Project one GraphQL row into the agent shape (snake keys, id→sqid, nesting).

_CompiledTool

class _CompiledTool(Tool)

A FastMCP tool that runs a GraphQL operation under the request actor.

Carries the introspection-derived execution plan as data (no closures) so the subclass stays a plain pydantic model; :meth:run interprets it.

run

async def run(arguments: dict[str, Any]) -> ToolResult

Execute the operation and return the projected payload as structured content.