Skip to content

ccproxy.llms.formatters.context

ccproxy.llms.formatters.context

Context helpers for formatter conversions using async contextvars.

register_request 

register_request(request, instructions=None)

Record the most recent upstream request for streaming conversions.

Source code in ccproxy/llms/formatters/context.py 
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
def register_request(request: Any | None, instructions: str | None = None) -> None:
    """Record the most recent upstream request for streaming conversions."""

    normalized = instructions.strip() if isinstance(instructions, str) else None

    _REQUEST_VAR.set(request)
    _INSTRUCTIONS_VAR.set(normalized)

    try:
        from ccproxy.core.request_context import RequestContext

        ctx = RequestContext.get_current()
        if ctx is not None:
            formatter_state = ctx.metadata.setdefault("formatter_state", {})
            if request is None:
                formatter_state.pop("request", None)
            else:
                formatter_state["request"] = request

            if normalized:
                formatter_state["instructions"] = normalized
            elif instructions is None:
                formatter_state.pop("instructions", None)
    except Exception:
        # Request context propagation is best-effort; proceed even when
        # request context is unavailable (e.g., during unit tests).
        pass

get_last_request 

get_last_request()

Return the cached upstream request for the active conversion, if any.

Source code in ccproxy/llms/formatters/context.py 
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
def get_last_request() -> Any | None:
    """Return the cached upstream request for the active conversion, if any."""

    try:
        from ccproxy.core.request_context import RequestContext

        ctx = RequestContext.get_current()
        if ctx is not None:
            formatter_state = ctx.metadata.get("formatter_state", {})
            if "request" in formatter_state:
                return formatter_state["request"]
    except Exception:
        pass

    return _REQUEST_VAR.get()

get_last_instructions 

get_last_instructions()

Return the cached instruction string from the last registered request.

Source code in ccproxy/llms/formatters/context.py 
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
def get_last_instructions() -> str | None:
    """Return the cached instruction string from the last registered request."""

    try:
        from ccproxy.core.request_context import RequestContext

        ctx = RequestContext.get_current()
        if ctx is not None:
            formatter_state = ctx.metadata.get("formatter_state", {})
            instructions = formatter_state.get("instructions")
            if isinstance(instructions, str) and instructions.strip():
                return instructions.strip()
    except Exception:
        pass

    return _INSTRUCTIONS_VAR.get()

register_request_tools 

register_request_tools(tools)

Cache request tool definitions for downstream streaming responses.

Source code in ccproxy/llms/formatters/context.py 
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
def register_request_tools(tools: list[Any] | None) -> None:
    """Cache request tool definitions for downstream streaming responses."""

    normalized = list(tools) if tools else None
    _TOOLS_VAR.set(normalized)

    try:
        from ccproxy.core.request_context import RequestContext

        ctx = RequestContext.get_current()
        if ctx is not None:
            formatter_state = ctx.metadata.setdefault("formatter_state", {})
            if normalized is None:
                formatter_state.pop("tools", None)
            else:
                formatter_state["tools"] = normalized
    except Exception:
        pass

get_last_request_tools 

get_last_request_tools()

Return cached request tool definitions, if any.

Source code in ccproxy/llms/formatters/context.py 
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
def get_last_request_tools() -> list[Any] | None:
    """Return cached request tool definitions, if any."""

    try:
        from ccproxy.core.request_context import RequestContext

        ctx = RequestContext.get_current()
        if ctx is not None:
            formatter_state = ctx.metadata.get("formatter_state", {})
            tools = formatter_state.get("tools")
            if isinstance(tools, list):
                return list(tools)
    except Exception:
        pass

    cached = _TOOLS_VAR.get()
    return list(cached) if cached else None