"""Explicit tool contracts. Every call is scoped to a workspace the grant covers: implied by a project_id, named by workspace_id, or the grant's only workspace.""" from uuid import UUID from pydantic import Field from src.core.exceptions import ForbiddenError, NotFoundError, ValidationError from src.domain.agent_access.asset_upload import confirm_upload, request_upload from src.domain.agent_access.operations import ( AttachmentInput, AgentOperationService, ConfirmAssetUpload, CreateProject, GetOperation, RequestAssetUpload, StartBuild, ) from src.domain.agent_access.rate_limit import ( READ_EXPENSIVE, READ_LIGHT, TOOL_RATE_CLASS, enforce_tool_rate_limit, ) from src.domain.agent_access.schemas import StrictModel from src.domain.billing.agent_purchases import ( AgentPurchaseService, CreditPurchaseRequest, ListCreditPackages, ) from src.domain.codegen.source_adoption import redact_source def _public_required_user_inputs(manifest) -> dict | None: """Expose manifest metadata only; submitted values never cross agent access.""" if not isinstance(manifest, dict) or not isinstance(manifest.get("items"), list): return None items = [] for item in manifest["items"]: if not isinstance(item, dict) or not item.get("id") or not item.get("label"): continue items.append( { "id": str(item["id"]), "label": str(item["label"]), "description": redact_source(str(item.get("description") or "")), "type": "secret" if item.get("type") == "secret" else "public", } ) return {"version": manifest.get("version", 1), "items": items} if items else None def _source_fidelity(run) -> dict | None: """How faithfully a run could follow the caller's source, when it was given any to convert. ``content_only`` means a migrate/replace import carried no UI markup, so the design was recreated from the prompt's description, not adopted from the caller's files. Returned to the calling agent so it can say so (and re-send the files) instead of presenting the result as a faithful conversion. None when the run had no source-adoption plan.""" ctx = (getattr(run, "plan", None) or {}).get("context") or {} plan = ctx.get("source_adoption_plan") if not isinstance(plan, dict) or plan.get("operation") not in ("migrate", "replace"): return None if plan.get("source_fidelity") != "content_only": return {"level": plan.get("source_fidelity") or "full"} return { "level": "content_only", "missing": list(plan.get("missing_source") or ["ui_markup"]), "notice": ( "No UI source (HTML/Vue/React markup) was attached, so this design was recreated from " "the description rather than adopted from your files. To keep your design, start a new " "build with every source file attached." ), } def _needs_owner_approval(run) -> bool: if getattr(run, "status", None) != "blocked": return False ctx = (getattr(run, "plan", None) or {}).get("context") or {} state = ctx.get("work_branch") or {} if state.get("failure_class") == "approval_required": return True # Without the work branch there is no stored state: the gate's own message says the same thing. return str(getattr(run, "error_message", None) or "").startswith("Waiting for approval:") class ListProjects(StrictModel): # Omit to list across every workspace the connection covers. workspace_id: UUID | None = None offset: int = Field(default=0, ge=0, le=100_000) limit: int = Field(default=25, ge=1, le=100) class GetProject(StrictModel): project_id: UUID class GetBuild(GetProject): run_id: UUID class EstimateBuild(GetProject): prompt: str = Field(min_length=1, max_length=32_000) class InspectSource(StrictModel): """Inspect source handed off by an external connector before a build starts.""" project_id: UUID attachments: list[AttachmentInput] = Field(default_factory=list, max_length=60) class ListBuilds(GetProject): offset: int = Field(default=0, ge=0, le=100_000) limit: int = Field(default=25, ge=1, le=100) # Which tools each scope unlocks (order is the order tools are advertised in). _SCOPE_TOOLS = ( ( "projects.read", ["list_projects", "get_project", "get_build", "list_builds", "estimate_build"], ), ("projects.create", ["create_project"]), ("builds.run", ["inspect_source", "start_build", "request_asset_upload", "confirm_asset_upload"]), ("billing.read", ["list_credit_packages"]), ("credits.checkout", ["create_credit_checkout"]), ("credits.purchase", ["purchase_credits"]), ) class AgentToolService: def __init__(self, access, repository): self.access = access self.repository = repository async def usable_workspaces(self, principal) -> tuple[list[str], list[dict]]: """(tools, workspaces) this token can use right now. A workspace where the authorizing person has lost a permission drops out; the rest keep working.""" usable: dict[str, set[UUID]] = {} tools: list[str] = [] for scope, names in _SCOPE_TOOLS: if scope not in principal.scopes: continue workspace_ids = await self.access.scoped_workspaces(principal, scope) if workspace_ids: usable[scope] = set(workspace_ids) tools += names if not usable: raise ForbiddenError("Agent scope not granted") if set(usable) & {"projects.create", "builds.run", "credits.checkout", "credits.purchase"}: tools.append("get_operation") workspaces = [] for row in await self.repository.grant_workspaces(principal.id): scopes = [scope for scope, ids in usable.items() if row.workspace_id in ids] if not scopes: continue summary = await self.access.workspaces.get_workspace_summary(row.workspace_id) workspaces.append( { "id": str(row.workspace_id), "name": summary.name if summary else "", "scopes": scopes, } ) if len(workspaces) > 1: # Single-workspace connections never need to choose, so they never see it. tools.insert(0, "list_workspaces") return tools, workspaces async def capabilities(self, principal): tools, workspaces = await self.usable_workspaces(principal) # `workspace_id` is the first workspace only; kept so an older MCP Worker that # still requires the key keeps parsing. Newer callers read `workspaces`. return {"workspace_id": workspaces[0]["id"], "workspaces": workspaces, "tools": tools} async def _rate_limit_workspace(self, principal, arguments) -> UUID | None: """The workspace a call targets, for rate-limit accounting only. Never trusts a caller-supplied id that is not part of this grant, so a connection cannot burn another tenant's workspace budget.""" if self.access.redis is None: return None # limiter disabled: nothing to attribute, skip the lookup granted = {row.workspace_id for row in await self.repository.grant_workspaces(principal.id)} try: if arguments.get("workspace_id"): candidate = UUID(str(arguments["workspace_id"])) elif arguments.get("project_id"): project = await self.access.projects.get_by_id(UUID(str(arguments["project_id"]))) candidate = project.workspace_id if project else None else: candidate = next(iter(granted)) if len(granted) == 1 else None except (ValueError, TypeError, AttributeError): return None return candidate if candidate in granted else None async def execute(self, principal, name: str, arguments: dict): # Mutation tools routed through AgentOperationService (write.standard / commerce) # rate-limit themselves after their idempotency-key cache-hit check, so a safe # retry never consumes or trips a limit. request_asset_upload/confirm_asset_upload # are write.standard too but have no idempotency key to gate on (see # asset_upload.py), so they're limited here up front like a read tool instead. if TOOL_RATE_CLASS.get(name) in (READ_LIGHT, READ_EXPENSIVE) or name in { "request_asset_upload", "confirm_asset_upload", }: await enforce_tool_rate_limit( self.access.redis, grant_id=principal.id, workspace_id=await self._rate_limit_workspace(principal, arguments), tool_name=name, ) if name == "list_workspaces": StrictModel.model_validate(arguments) _, workspaces = await self.usable_workspaces(principal) return {"workspaces": workspaces} if name == "estimate_build": from src.domain.codegen.credit_estimator import estimate_codegen_credits body = EstimateBuild.model_validate(arguments) await self.access.require_scope(principal, "projects.read", body.project_id) return estimate_codegen_credits(prompt=body.prompt).__dict__ if name == "inspect_source": from src.domain.adopt.import_scan import scan_import body = InspectSource.model_validate(arguments) await self.access.require_scope(principal, "builds.run", body.project_id) if any(attachment.asset_id is not None for attachment in body.attachments): raise ValidationError( "inspect_source needs inline source content; use confirm_asset_upload and start_build for asset-backed files" ) files = {attachment.path: attachment.content for attachment in body.attachments} analysis = scan_import(files) return { "project_id": str(body.project_id), "files": [{"path": path, "size": len(content or "")} for path, content in files.items()], "adoption_analysis": analysis, } if name == "list_builds": body = ListBuilds.model_validate(arguments) workspace_id = await self.access.require_scope( principal, "projects.read", body.project_id ) rows = await self.repository.visible_runs( workspace_id, body.project_id, body.offset, body.limit + 1 ) return { "builds": [ {"run_id": str(row.id), "status": row.status} for row in rows[: body.limit] ], "next_offset": body.offset + body.limit if len(rows) > body.limit else None, } if name == "list_credit_packages": body = ListCreditPackages.model_validate(arguments) return await AgentPurchaseService(self.access).catalog(principal, body.workspace_id) if name in {"create_credit_checkout", "purchase_credits"}: body = CreditPurchaseRequest.model_validate(arguments) return await AgentPurchaseService(self.access).submit(principal, name, body) if name == "get_build": body = GetBuild.model_validate(arguments) workspace_id = await self.access.require_scope( principal, "projects.read", body.project_id ) run = await self.repository.visible_run(workspace_id, body.project_id, body.run_id) if not run: raise NotFoundError("Build not found") return { "run_id": str(run.id), "project_id": str(run.project_id), "status": run.status, "revision_id": str(run.revision_id) if run.revision_id else None, "result_summary": ( redact_source(run.result_summary) if run.result_summary else None ), # Why a run did not finish (see classify_provisioning_failure). Redacted # like the summary, and capped so a long runner error cannot bloat this. "failure_message": ( redact_source(run.error_message)[:600] if run.status in {"failed", "blocked"} and run.error_message else None ), # True once the run's work is saved on its own branch, so a failure did not lose it # and a retry resumes instead of regenerating (see work_branch.py). # "content_only": the import had no UI markup, so the design was recreated from the # prompt's description rather than adopted from the caller's files. "source_fidelity": _source_fidelity(run), # True while the build waits for the project owner to approve security-sensitive # capabilities (public access, deletes, email...). Only a workspace admin can decide, # in the dashboard: retrying before that changes nothing. "needs_owner_approval": _needs_owner_approval(run), "work_saved": bool( ( ((getattr(run, "plan", None) or {}).get("context") or {}).get("work_branch") or {} ).get("last_checkpoint") ), "required_user_inputs": ( _public_required_user_inputs(run.pending_user_inputs) if run.status == "user_input_required" else None ), "dashboard_url": ( f"{self.access.settings.app_url.rstrip('/')}/projects/{run.project_id}" ), } if name == "request_asset_upload": body = RequestAssetUpload.model_validate(arguments) return await request_upload(self.access, principal, body) if name == "confirm_asset_upload": body = ConfirmAssetUpload.model_validate(arguments) return await confirm_upload(self.access, principal, body) if name in {"create_project", "start_build"}: body = (CreateProject if name == "create_project" else StartBuild).model_validate( arguments ) if name == "start_build": body.ensure_ui_source() return await AgentOperationService(self.access).submit(principal, name, body) if name == "get_operation": body = GetOperation.model_validate(arguments) return await AgentOperationService(self.access).get(principal, body.request_key) if name == "list_projects": body = ListProjects.model_validate(arguments) if body.workspace_id is not None: await self.access.require_scope( principal, "projects.read", workspace_id=body.workspace_id ) usable = {body.workspace_id} else: usable = set(await self.access.scoped_workspaces(principal, "projects.read")) if not usable: raise ForbiddenError("Agent scope not granted") access_rows = [ row for row in await self.repository.grant_workspaces(principal.id) if row.workspace_id in usable ] rows = await self.repository.visible_projects(access_rows, body.offset, body.limit + 1) return { "projects": [self.project_view(row) for row in rows[: body.limit]], "next_offset": body.offset + body.limit if len(rows) > body.limit else None, } if name == "get_project": body = GetProject.model_validate(arguments) workspace_id = await self.access.require_scope( principal, "projects.read", body.project_id ) row = await self.repository.visible_project(workspace_id, body.project_id) if not row: raise NotFoundError("Project not found") return self.project_view(row) raise ValidationError("Unknown or unavailable agent tool") @staticmethod def project_view(project): # Never serialize ORM objects, repo credentials, environment values, or internal URLs. return { "id": str(project.id), "workspace_id": str(project.workspace_id), "name": project.name, "type": project.type, "status": project.status, "active_revision_id": str(project.active_revision_id) if project.active_revision_id else None, }