Composite crew-dcs workflows exposed as single MCP tools.
These read like the agent's intent ("roles and the users assigned to each") and
fan out server-side over one pooled HTTP session, returning a structured summary
the model reads directly.
get_roles_with_membership
async
get_roles_with_membership(
auth: Any, *, context: RouteContext | None = None
) -> dict
Return every role in the instance together with the users assigned to it.
Source code in src/crew_dcs/mcp_server/workflows.py
19
20
21
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
48 | async def get_roles_with_membership(
auth: Any,
*,
context: RouteContext | None = None,
) -> dict:
"""Return every role in the instance together with the users assigned to it."""
context = context or RouteContext()
owns_session = context.session is None
if owns_session:
context.session = httpx.AsyncClient()
try:
roles = await DomoRoles(auth=auth).get(context=context)
summary = []
for role in roles:
members = await role.get_membership(context=context)
summary.append(
{
"id": role.id,
"name": role.name,
"member_count": len(members),
"members": [
{"id": m.id, "display_name": m.display_name} for m in members
],
}
)
return {"roles": summary}
finally:
if owns_session and context.session is not None:
await context.session.aclose()
context.session = None
|
update_users_role
async
update_users_role(
auth: Any,
user_ids: list[str],
target_role_id: str,
dry_run: bool = True,
*,
ctx: Any = None,
context: RouteContext | None = None
) -> dict
Assign users to a role. Defaults to dry_run: returns the plan without
applying. Call again with dry_run=False to apply (per-item, fault-tolerant).
Source code in src/crew_dcs/mcp_server/workflows.py
51
52
53
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
79
80
81
82
83
84
85
86 | async def update_users_role(
auth: Any,
user_ids: list[str],
target_role_id: str,
dry_run: bool = True,
*,
ctx: Any = None,
context: RouteContext | None = None,
) -> dict:
"""Assign users to a role. Defaults to dry_run: returns the plan without
applying. Call again with dry_run=False to apply (per-item, fault-tolerant)."""
if dry_run:
return {
"dry_run": True,
"targets": [
{"user_id": uid, "proposed_role_id": target_role_id} for uid in user_ids
],
}
results = []
total = len(user_ids)
for index, uid in enumerate(user_ids):
try:
await role_routes.role_membership_add_users(
auth=auth,
role_id=target_role_id,
user_ids=[uid],
context=context,
)
results.append({"user_id": uid, "success": True, "error": None})
except Exception as exc: # noqa: BLE001
results.append({"user_id": uid, "success": False, "error": str(exc)})
if ctx is not None:
await ctx.report_progress(index + 1, total)
await ctx.info(f"role {target_role_id}: processed user {uid}")
return {"dry_run": False, "results": results}
|