cc_agent_login

Instructions

Implementation Reference

• src/opensips_mcp/tools/call_center_tools.py:32-54 (handler)

  The MCP tool handler for cc_agent_login. Decorated with @mcp.tool() for registration, @audited for audit logging, and @require_permission("mi.write") for RBAC. Accepts agent_id (str) and state (int: 1=login, 0=logout), then dispatches the command via the MI client.

  @mcp.tool()
  @audited("cc_agent_login")
  @require_permission("mi.write")
  async def cc_agent_login(
      ctx: Context,
      agent_id: str,
      state: int,
  ) -> dict[str, Any]:
      """Log a call center agent in or out.
  
      Parameters
      ----------
      agent_id:
          The agent identifier (e.g. ``agent1``).
      state:
          Login state: 1 to log in, 0 to log out.
      """
      app = ctx.request_context.lifespan_context
      result = await app.mi_client.execute(
          "cc_agent_login",
          {"agent_id": agent_id, "state": str(state)},
      )
      return result

• src/opensips_mcp/server.py:138-138 (registration)

  The tool module call_center_tools (containing cc_agent_login) is registered by importing it in server.py, which triggers the @mcp.tool() decorator and registers the handler with the FastMCP server instance.

  mcp = FastMCP("opensips-mcp-server", lifespan=app_lifespan)

• src/opensips_mcp/mi/commands.py:153-153 (schema)

  MI command registry entry for cc_agent_login, defining its module (call_center), description, parameters (agent_id, state), permission (mi.write), and category.

  _r("cc_agent_login", "call_center", "Log a call center agent in or out", ["agent_id", "state"], "mi.write", "call_center")

• src/opensips_mcp/security/audit.py:45-63 (helper)

  The @audited decorator applied to cc_agent_login, which logs audit entries (with operation name, parameters, role, and success/error status) on each invocation.

  def audited(operation: str):
      """Decorator that logs audit entries for tool calls."""
  
      def decorator(func):
          @wraps(func)
          async def wrapper(ctx: Context, *args, **kwargs):
              app = ctx.request_context.lifespan_context
              role = getattr(app.settings, "role", "unknown")
              try:
                  result = await func(ctx, *args, **kwargs)
                  audit_log(operation, kwargs, "success", role)
                  return result
              except Exception as e:
                  audit_log(operation, kwargs, f"error: {e}", role)
                  raise
  
          return wrapper
  
      return decorator

• src/opensips_mcp/security/rbac.py:93-132 (helper)

  The @require_permission decorator applied to cc_agent_login, which enforces that the active role has the 'mi.write' permission before allowing execution.

  def require_permission(permission: str) -> Callable[..., Any]:
      """Decorator that enforces RBAC on an MCP tool function.
  
      The decorated function **must** accept ``ctx: Context`` as its first
      positional argument.  The current role is read from
      ``ctx.request_context.lifespan_context.settings.role``.
  
      Raises ``McpError`` when the active role lacks the requested permission.
      """
  
      def decorator(func: Callable[..., Any]) -> Callable[..., Any]:
          @functools.wraps(func)
          async def wrapper(ctx: Context, *args: Any, **kwargs: Any) -> Any:
              app_context = ctx.request_context.lifespan_context
              role_value: str = app_context.settings.role.lower()
  
              try:
                  role = Role(role_value)
              except ValueError:
                  raise McpError(
                      ErrorData(
                          code=-1,
                          message=f"Unknown role '{role_value}'. Valid roles: {[r.value for r in Role]}",
                      )
                  ) from None
  
              allowed = PERMISSIONS.get(role, set())
              if permission not in allowed:
                  raise McpError(
                      ErrorData(
                          code=-1,
                          message=f"Permission denied: role '{role.value}' lacks '{permission}' permission",
                      )
                  )
  
              return await func(ctx, *args, **kwargs)
  
          return wrapper
  
      return decorator