Build a Streaming Markdown Code Fence Parser

Streaming Markdown Code Fence Parser

In generative AI interfaces (such as Cursor), text streams in token by token. To render syntax-highlighted code blocks in real time, the client must detect when a code block begins, streams content, and closes—even when fence delimiters are split across consecutive chunks.

Implement a class StreamingCodeFenceParser that processes streaming text chunks and yields structured events.

Markdown Code Fence Specifications

1. Opening Fence:
   • Starts at the beginning of the stream or immediately after a newline (\n).
   • May be preceded by 0 to 3 spaces (which are ignored).
   • Consists of a run of 3 or more consecutive backticks (`) or tildes (~).
   • The delimiter character ( or~`) cannot be mixed.
   • Any remaining text on the same line after the fence (excluding leading/trailing whitespace) is the info string / language tag.
   • For backtick fences, the info string cannot contain any backtick characters.
2. Code Content:
   • All characters following the opening fence's newline up until the start of the closing fence.
3. Closing Fence:
   • Starts on a new line, optionally preceded by 0 to 3 spaces.
   • Consists of the same delimiter character as the opening fence.
   • Must have at least as many delimiter characters as the opening fence.
   • May only be followed by optional spaces and a newline (\n) or the end of the stream (no info string allowed).
4. Normal Text:
   • Any content outside of a code block.

Class Interface

class StreamingCodeFenceParser:
    def __init__(self):
        pass

    def feed(self, chunk: str) -> list[dict]:
        """
        Process an incoming chunk of text and return zero or more events.
        """
        pass

    def flush(self) -> list[dict]:
        """
        Signal the end of the stream. Flush any buffered content as text
        or unclosed code content, returning final events.
        """
        pass

Event Format

Emit events as dictionaries:

• {"type": "text", "content": str}: Emitted for non-code text as soon as it is confirmed not to be part of a fence.
• {"type": "code_start", "language": str}: Emitted when an opening fence line is fully resolved (language is empty string "" if unspecified).
• {"type": "code_chunk", "content": str}: Emitted for content inside the code block as it becomes available.
• {"type": "code_end"}: Emitted when a valid closing fence is confirmed.

Constraints

• Chunks can be of arbitrary length (from single characters to full documents).
• A closing fence cannot close a block opened with a different character (e.g., ~~~ cannot close ```).
• Content should be emitted as early as possible without prematurely committing to a state that could change with subsequent characters.

Example 1

parser = StreamingCodeFenceParser()

parser.feed("Here is code:\n```py")
# Returns: [{"type": "text", "content": "Here is code:\n"}]
# ("```py" is buffered because the opening fence line is not complete yet)

parser.feed("thon\nprint(1)\n```\nDone")
# Returns:
# [
#   {"type": "code_start", "language": "python"},
#   {"type": "code_chunk", "content": "print(1)\n"},
#   {"type": "code_end"},
#   {"type": "text", "content": "Done"}
# ]

Example 2 (Delimiter Split Across Chunks)

parser = StreamingCodeFenceParser()

parser.feed("``")
# Returns: [] (could be start of a fence or plain text)

parser.feed("`\nhello\n```")
# Returns:
# [
#   {"type": "code_start", "language": ""},
#   {"type": "code_chunk", "content": "hello\n"}
# ]
# (Trailing "```" is buffered because it might be extended or followed by text)

parser.flush()
# "```" at end of stream acts as valid closing fence
# Returns: [{"type": "code_end"}]