acc_query_cdrs
Retrieve call detail records (CDRs) from accounting data by specifying optional date range, caller, or callee filters.
Instructions
Query accounting CDRs with optional date range and caller/callee filters.
Parameters
date_from:
ISO-format start date/time filter (e.g. 2025-01-01T00:00:00).
date_to:
ISO-format end date/time filter.
caller:
Partial match on the caller field.
callee:
Partial match on the callee field.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| date_from | No | ||
| date_to | No | ||
| caller | No | ||
| callee | No | ||
| limit | No | ||
| offset | No |
Output Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
No arguments | |||
Implementation Reference
- The `acc_query_cdrs` async function is the actual handler for querying accounting CDRs. It accepts optional date_from, date_to, caller, callee filters plus limit/offset, builds a SQLAlchemy query using the helper `_build_acc_query`, executes it via the DB session, serializes results with `_serialize_acc`, and returns a dict with 'cdrs' and 'count'.
@mcp.tool() @require_permission("db.read") async def acc_query_cdrs( ctx: Context, date_from: str | None = None, date_to: str | None = None, caller: str | None = None, callee: str | None = None, limit: int = 100, offset: int = 0, ) -> dict[str, Any]: """Query accounting CDRs with optional date range and caller/callee filters. Parameters ---------- date_from: ISO-format start date/time filter (e.g. ``2025-01-01T00:00:00``). date_to: ISO-format end date/time filter. caller: Partial match on the caller field. callee: Partial match on the callee field. """ from opensips_mcp.db.models.acc import Acc app = ctx.request_context.lifespan_context stmt = _build_acc_query(Acc, date_from, date_to, caller, callee, limit, offset) async with app.db_session_factory() as session: result = await session.execute(stmt) rows = list(result.scalars().all()) return { "cdrs": [_serialize_acc(r) for r in rows], "count": len(rows), } - The `Acc` SQLAlchemy model defines the schema for the `acc` table columns (id, method, from_tag, to_tag, callid, sip_code, sip_reason, time, duration, setuptime, caller, callee, etc.) which map to the fields returned by the tool.
class Acc(Base): __tablename__ = "acc" id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True) method: Mapped[str] = mapped_column(String(16), default="") from_tag: Mapped[str] = mapped_column(String(64), default="") to_tag: Mapped[str] = mapped_column(String(64), default="") callid: Mapped[str] = mapped_column(String(255), default="") sip_code: Mapped[str] = mapped_column(String(3), default="") sip_reason: Mapped[str] = mapped_column(String(128), default="") time: Mapped[datetime] = mapped_column(DateTime, nullable=False) duration: Mapped[int] = mapped_column(Integer, default=0) setuptime: Mapped[int] = mapped_column(Integer, default=0) created: Mapped[datetime] = mapped_column(DateTime, nullable=True) caller: Mapped[str] = mapped_column(String(255), default="") callee: Mapped[str] = mapped_column(String(255), default="") class MissedCall(Base): __tablename__ = "missed_calls" id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True) method: Mapped[str] = mapped_column(String(16), default="") from_tag: Mapped[str] = mapped_column(String(64), default="") to_tag: Mapped[str] = mapped_column(String(64), default="") callid: Mapped[str] = mapped_column(String(255), default="") sip_code: Mapped[str] = mapped_column(String(3), default="") sip_reason: Mapped[str] = mapped_column(String(128), default="") time: Mapped[datetime] = mapped_column(DateTime, nullable=False) duration: Mapped[int] = mapped_column(Integer, default=0) setuptime: Mapped[int] = mapped_column(Integer, default=0) created: Mapped[datetime] = mapped_column(DateTime, nullable=True) caller: Mapped[str] = mapped_column(String(255), default="") callee: Mapped[str] = mapped_column(String(255), default="") - src/opensips_mcp/server.py:161-161 (registration)The `acc_tools` module is imported in `server.py`, which triggers the `@mcp.tool()` decorator on `acc_query_cdrs` and registers it as an MCP tool.
from opensips_mcp.tools import acc_tools as _acc_tools # noqa: E402, F401 - `_build_acc_query` is a helper that constructs a filtered SQLAlchemy SELECT query using the model, applying optional date_from, date_to, caller, and callee filters, with ordering by time descending and limit/offset.
def _build_acc_query(model: Any, date_from: str | None, date_to: str | None, caller: str | None, callee: str | None, limit: int, offset: int): """Build a filtered query for an accounting table.""" stmt = select(model) if date_from: stmt = stmt.where(model.time >= datetime.fromisoformat(date_from)) if date_to: stmt = stmt.where(model.time <= datetime.fromisoformat(date_to)) if caller: stmt = stmt.where(model.caller.contains(caller)) if callee: stmt = stmt.where(model.callee.contains(callee)) stmt = stmt.order_by(model.time.desc()).limit(limit).offset(offset) return stmt - `_serialize_acc` is a helper that converts an Acc (or MissedCall) SQLAlchemy row into a plain dict with ISO-formatted time string.
def _serialize_acc(row: Any) -> dict[str, Any]: """Serialize an Acc or MissedCall row to a dict.""" return { "id": row.id, "method": row.method, "from_tag": row.from_tag, "to_tag": row.to_tag, "callid": row.callid, "sip_code": row.sip_code, "sip_reason": row.sip_reason, "time": row.time.isoformat() if row.time else None, "duration": row.duration, "setuptime": row.setuptime, "caller": row.caller, "callee": row.callee, }