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 
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
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 
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
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 
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
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 
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
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 
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
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

register_openai_thinking_xml 

register_openai_thinking_xml(enabled)

Cache OpenAI thinking serialization preference for active conversions.

Parameters:

Name Type Description Default
enabled bool | None

Whether thinking blocks should be serialized with XML wrappers. None means downstream conversion logic should use its default.

 required
Note

The value is stored in a ContextVar, so concurrent async requests keep independent preferences without leaking into each other.

Source code in ccproxy/llms/formatters/context.py 
122
123
124
125
126
127
128
129
130
131
132
133
134
def register_openai_thinking_xml(enabled: bool | None) -> None:
    """Cache OpenAI thinking serialization preference for active conversions.

    Args:
        enabled: Whether thinking blocks should be serialized with XML wrappers.
            ``None`` means downstream conversion logic should use its default.

    Note:
        The value is stored in a ``ContextVar``, so concurrent async requests
        keep independent preferences without leaking into each other.
    """

    _OPENAI_THINKING_XML_VAR.set(enabled)

get_openai_thinking_xml 

get_openai_thinking_xml()

Return the OpenAI thinking serialization preference for active conversions.

Source code in ccproxy/llms/formatters/context.py 
137
138
139
140
def get_openai_thinking_xml() -> bool | None:
    """Return the OpenAI thinking serialization preference for active conversions."""

    return _OPENAI_THINKING_XML_VAR.get()